notifyr: ntfy send/read client for agents
send (title/priority/tags), read (poll mode, greppable one-liners), topics (known homelab buses from config). Config with homelab defaults on first run. Unit tests against httptest; smoked against the real ntfy on topic agent-tools-test. Spec: doc/tool-parity.md 3.3. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011KikHkfCiC3yELbsMN8fT9
This commit is contained in:
78
notifyr/notify/config.go
Normal file
78
notifyr/notify/config.go
Normal file
@@ -0,0 +1,78 @@
|
||||
package notify
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
// Config is ~/.config/notifyr/config.json, created with homelab
|
||||
// defaults on first run. NOTIFYR_CONFIG overrides the path.
|
||||
type Config struct {
|
||||
Server string `json:"server"`
|
||||
Token string `json:"token"`
|
||||
DefaultTopic string `json:"default_topic"`
|
||||
Topics map[string]string `json:"topics"` // known topics -> what they carry (informational)
|
||||
}
|
||||
|
||||
func DefaultConfig() *Config {
|
||||
return &Config{
|
||||
Server: "https://ntfy.brasse-pc.eu",
|
||||
Token: "",
|
||||
DefaultTopic: "claude",
|
||||
Topics: map[string]string{
|
||||
"claude": "agents' direct notes to Björn",
|
||||
"agent-helm": "agent-helm events (question waiting, session died)",
|
||||
"Info": "*arr system events",
|
||||
"media-hamtningar": "media grabbed for download",
|
||||
"media-nytt": "new media landed in Jellyfin",
|
||||
"pi5-server": "server maintenance (reboots, watchtower)",
|
||||
"pi5-server-fel": "server problems: failed units, disk space",
|
||||
"ci-fel": "failed Gitea Actions builds",
|
||||
"monitoring": "Uptime Kuma up/down alerts",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func ConfigPath() string {
|
||||
if p := os.Getenv("NOTIFYR_CONFIG"); p != "" {
|
||||
return p
|
||||
}
|
||||
dir, err := os.UserConfigDir()
|
||||
if err != nil {
|
||||
dir = "."
|
||||
}
|
||||
return filepath.Join(dir, "notifyr", "config.json")
|
||||
}
|
||||
|
||||
// LoadConfig reads the config, creating it with defaults on first run.
|
||||
func LoadConfig(path string) (*Config, error) {
|
||||
cfg := DefaultConfig()
|
||||
data, err := os.ReadFile(path)
|
||||
if os.IsNotExist(err) {
|
||||
if err := SaveConfig(path, cfg); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
fmt.Fprintf(os.Stderr, "notifyr: created %s\n", path)
|
||||
return cfg, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := json.Unmarshal(data, cfg); err != nil {
|
||||
return nil, fmt.Errorf("%s: %w", path, err)
|
||||
}
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
func SaveConfig(path string, cfg *Config) error {
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
data, err := json.MarshalIndent(cfg, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, append(data, '\n'), 0o600)
|
||||
}
|
||||
165
notifyr/notify/notify.go
Normal file
165
notifyr/notify/notify.go
Normal file
@@ -0,0 +1,165 @@
|
||||
// Package notify is a thin client for a ntfy server: publish
|
||||
// notifications and poll past ones, so agents can both alert humans
|
||||
// and check what the infrastructure has been complaining about.
|
||||
package notify
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Message is one ntfy message as returned by the /json poll endpoint.
|
||||
type Message struct {
|
||||
ID string `json:"id"`
|
||||
Time int64 `json:"time"`
|
||||
Event string `json:"event"`
|
||||
Topic string `json:"topic"`
|
||||
Title string `json:"title"`
|
||||
Message string `json:"message"`
|
||||
Priority int `json:"priority"`
|
||||
Tags []string `json:"tags"`
|
||||
}
|
||||
|
||||
// Client talks to one ntfy server.
|
||||
type Client struct {
|
||||
Server string // e.g. https://ntfy.brasse-pc.eu
|
||||
Token string // optional bearer token
|
||||
HTTP *http.Client
|
||||
}
|
||||
|
||||
func New(server, token string) *Client {
|
||||
return &Client{
|
||||
Server: strings.TrimRight(server, "/"),
|
||||
Token: token,
|
||||
HTTP: &http.Client{Timeout: 30 * time.Second},
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) auth(req *http.Request) {
|
||||
if c.Token != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+c.Token)
|
||||
}
|
||||
}
|
||||
|
||||
// ValidPriority reports whether p is a priority ntfy accepts.
|
||||
func ValidPriority(p string) bool {
|
||||
switch p {
|
||||
case "", "1", "2", "3", "4", "5", "min", "low", "default", "high", "max", "urgent":
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Send publishes a message to a topic.
|
||||
func (c *Client) Send(topic, title, msg, priority string, tags []string) error {
|
||||
if topic == "" {
|
||||
return fmt.Errorf("no topic given (flag --topic or default_topic in the config)")
|
||||
}
|
||||
if msg == "" {
|
||||
return fmt.Errorf("empty message")
|
||||
}
|
||||
if !ValidPriority(priority) {
|
||||
return fmt.Errorf("invalid priority %q (use min|low|default|high|urgent or 1-5)", priority)
|
||||
}
|
||||
req, err := http.NewRequest("POST", c.Server+"/"+url.PathEscape(topic), strings.NewReader(msg))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
c.auth(req)
|
||||
if title != "" {
|
||||
req.Header.Set("Title", title)
|
||||
}
|
||||
if priority != "" {
|
||||
req.Header.Set("Priority", priority)
|
||||
}
|
||||
if len(tags) > 0 {
|
||||
req.Header.Set("Tags", strings.Join(tags, ","))
|
||||
}
|
||||
resp, err := c.HTTP.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
|
||||
if resp.StatusCode >= 300 {
|
||||
return fmt.Errorf("server answered %s: %s", resp.Status, strings.TrimSpace(string(body)))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Read polls past messages from a topic. since accepts ntfy's formats:
|
||||
// a duration ("10m", "2h"), a unix timestamp, a message id, or "all".
|
||||
func (c *Client) Read(topic, since string, limit int) ([]Message, error) {
|
||||
if topic == "" {
|
||||
return nil, fmt.Errorf("no topic given (flag --topic or default_topic in the config)")
|
||||
}
|
||||
if since == "" {
|
||||
since = "all"
|
||||
}
|
||||
u := fmt.Sprintf("%s/%s/json?poll=1&since=%s", c.Server, url.PathEscape(topic), url.QueryEscape(since))
|
||||
req, err := http.NewRequest("GET", u, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.auth(req)
|
||||
resp, err := c.HTTP.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode >= 300 {
|
||||
body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
|
||||
return nil, fmt.Errorf("server answered %s: %s", resp.Status, strings.TrimSpace(string(body)))
|
||||
}
|
||||
var out []Message
|
||||
sc := bufio.NewScanner(resp.Body)
|
||||
sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||
for sc.Scan() {
|
||||
line := strings.TrimSpace(sc.Text())
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
var m Message
|
||||
if err := json.Unmarshal([]byte(line), &m); err != nil {
|
||||
continue // tolerate junk lines; poll output is one JSON object per line
|
||||
}
|
||||
if m.Event != "message" {
|
||||
continue
|
||||
}
|
||||
out = append(out, m)
|
||||
}
|
||||
if err := sc.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if limit > 0 && len(out) > limit {
|
||||
out = out[len(out)-limit:] // keep the newest
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Format renders a message as one stable, greppable line.
|
||||
func Format(m Message) string {
|
||||
ts := time.Unix(m.Time, 0).Format("2006-01-02 15:04:05")
|
||||
prio := ""
|
||||
switch {
|
||||
case m.Priority >= 4:
|
||||
prio = " [high]"
|
||||
case m.Priority > 0 && m.Priority <= 2:
|
||||
prio = " [low]"
|
||||
}
|
||||
title := ""
|
||||
if m.Title != "" {
|
||||
title = " (" + m.Title + ")"
|
||||
}
|
||||
tags := ""
|
||||
if len(m.Tags) > 0 {
|
||||
tags = " #" + strings.Join(m.Tags, " #")
|
||||
}
|
||||
return fmt.Sprintf("%s%s%s %s%s", ts, prio, title, strings.ReplaceAll(m.Message, "\n", " ⏎ "), tags)
|
||||
}
|
||||
135
notifyr/notify/notify_test.go
Normal file
135
notifyr/notify/notify_test.go
Normal file
@@ -0,0 +1,135 @@
|
||||
package notify
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestSendSetsHeadersAndBody(t *testing.T) {
|
||||
var got *http.Request
|
||||
var body string
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
got = r
|
||||
b := make([]byte, 1024)
|
||||
n, _ := r.Body.Read(b)
|
||||
body = string(b[:n])
|
||||
fmt.Fprint(w, `{"id":"x"}`)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := New(srv.URL, "tok123")
|
||||
err := c.Send("ci-fel", "Build failed", "arm64 test: FAIL", "high", []string{"warning", "ci"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.URL.Path != "/ci-fel" {
|
||||
t.Errorf("path = %q", got.URL.Path)
|
||||
}
|
||||
if body != "arm64 test: FAIL" {
|
||||
t.Errorf("body = %q", body)
|
||||
}
|
||||
for hdr, want := range map[string]string{
|
||||
"Title": "Build failed",
|
||||
"Priority": "high",
|
||||
"Tags": "warning,ci",
|
||||
"Authorization": "Bearer tok123",
|
||||
} {
|
||||
if v := got.Header.Get(hdr); v != want {
|
||||
t.Errorf("%s = %q, want %q", hdr, v, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendValidation(t *testing.T) {
|
||||
c := New("http://example.invalid", "")
|
||||
if err := c.Send("", "", "hello", "", nil); err == nil {
|
||||
t.Error("empty topic accepted")
|
||||
}
|
||||
if err := c.Send("t", "", "", "", nil); err == nil {
|
||||
t.Error("empty message accepted")
|
||||
}
|
||||
if err := c.Send("t", "", "hello", "banana", nil); err == nil {
|
||||
t.Error("bogus priority accepted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadParsesPollOutput(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Query().Get("poll") != "1" {
|
||||
t.Errorf("poll param missing: %s", r.URL.RawQuery)
|
||||
}
|
||||
if r.URL.Query().Get("since") != "2h" {
|
||||
t.Errorf("since = %q", r.URL.Query().Get("since"))
|
||||
}
|
||||
fmt.Fprintln(w, `{"id":"a","time":1754500000,"event":"message","topic":"t","message":"first","priority":3}`)
|
||||
fmt.Fprintln(w, `{"id":"b","time":1754500060,"event":"keepalive","topic":"t"}`)
|
||||
fmt.Fprintln(w, `not json at all`)
|
||||
fmt.Fprintln(w, `{"id":"c","time":1754500120,"event":"message","topic":"t","title":"T","message":"second","priority":5,"tags":["x"]}`)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
msgs, err := New(srv.URL, "").Read("t", "2h", 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(msgs) != 2 {
|
||||
t.Fatalf("got %d messages, want 2 (keepalive + junk filtered)", len(msgs))
|
||||
}
|
||||
if msgs[0].Message != "first" || msgs[1].Title != "T" {
|
||||
t.Errorf("unexpected messages: %+v", msgs)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadLimitKeepsNewest(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
for i := 1; i <= 5; i++ {
|
||||
fmt.Fprintf(w, "{\"id\":\"%d\",\"time\":%d,\"event\":\"message\",\"topic\":\"t\",\"message\":\"m%d\"}\n", i, 1754500000+i, i)
|
||||
}
|
||||
}))
|
||||
defer srv.Close()
|
||||
msgs, err := New(srv.URL, "").Read("t", "all", 2)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(msgs) != 2 || msgs[0].Message != "m4" || msgs[1].Message != "m5" {
|
||||
t.Errorf("limit should keep the newest: %+v", msgs)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFormatIsOneGreppableLine(t *testing.T) {
|
||||
line := Format(Message{Time: 1754500000, Title: "Backup", Message: "done\nall good", Priority: 4, Tags: []string{"ok"}})
|
||||
if strings.Contains(line, "\n") {
|
||||
t.Error("format must be a single line")
|
||||
}
|
||||
for _, want := range []string{"[high]", "(Backup)", "done", "#ok"} {
|
||||
if !strings.Contains(line, want) {
|
||||
t.Errorf("line %q missing %q", line, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfigRoundtrip(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "config.json")
|
||||
cfg, err := LoadConfig(path) // first run creates defaults
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cfg.Server == "" || cfg.DefaultTopic == "" {
|
||||
t.Error("defaults incomplete")
|
||||
}
|
||||
cfg.DefaultTopic = "elsewhere"
|
||||
if err := SaveConfig(path, cfg); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
again, err := LoadConfig(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if again.DefaultTopic != "elsewhere" {
|
||||
t.Error("saved change did not persist")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user