Files
VeillaxandClaude Sonnet 5.5 955e01333d Make the topic feeds work with the ntfy apps
The JSON stream now carries the event field the apps look for, sends open and keepalive
events, and supports ?poll=1 and ?since= with a duration or a message id, which is how the
apps catch up. Adds a walkthrough for the phone app to the README.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-27 04:22:53 -06:00

601 lines
19 KiB
Go

package sandbar
import (
"bufio"
"context"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"git.veillax.com/Veillax/reef/secretbox"
)
type fakeClock struct {
mu sync.Mutex
t time.Time
}
func (c *fakeClock) Now() time.Time { c.mu.Lock(); defer c.mu.Unlock(); return c.t }
func (c *fakeClock) Advance(d time.Duration) {
c.mu.Lock()
c.t = c.t.Add(d)
c.mu.Unlock()
}
type mailLog struct {
mu sync.Mutex
sent []string
}
func (m *mailLog) send(to, subject, body string) error {
m.mu.Lock()
m.sent = append(m.sent, to)
m.mu.Unlock()
return nil
}
func (m *mailLog) to() []string {
m.mu.Lock()
defer m.mu.Unlock()
return append([]string(nil), m.sent...)
}
func newTestServer(t *testing.T) (*Server, *fakeClock, *mailLog) {
t.Helper()
box, _ := secretbox.New(secretbox.GenerateKey())
st, err := OpenStore(t.TempDir(), box)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { st.Close() })
clk := &fakeClock{t: time.Date(2026, 1, 5, 9, 0, 0, 0, time.UTC)}
ml := &mailLog{}
return New(Options{Now: clk.Now, Mailer: ml.send, URL: "http://sandbar.test"}, st), clk, ml
}
func mustUser(t *testing.T, s *Server, name string) *User {
t.Helper()
u, err := s.createUser(name, strings.ToLower(name)+"@example.com", "correct-horse-battery", "member", "")
if err != nil {
t.Fatal(err)
}
return u
}
func TestScheduleRotationAndOverride(t *testing.T) {
start := time.Date(2026, 1, 5, 0, 0, 0, 0, time.UTC)
sc := &Schedule{Members: []string{"a", "b", "c"}, RotationHours: 24, Start: start.Unix()}
for day, want := range []string{"a", "b", "c", "a"} {
if got := sc.OnCall(start.Add(time.Duration(day)*24*time.Hour + time.Hour)); got != want {
t.Errorf("day %d: on call %q, want %q", day, got, want)
}
}
sc.Overrides = []Override{{UserID: "z", From: start.Add(2 * time.Hour).Unix(), To: start.Add(5 * time.Hour).Unix()}}
if got := sc.OnCall(start.Add(3 * time.Hour)); got != "z" {
t.Errorf("override: %q", got)
}
if got := sc.OnCall(start.Add(6 * time.Hour)); got != "a" {
t.Errorf("after override: %q", got)
}
if (&Schedule{}).OnCall(start) != "" {
t.Error("empty schedule should have nobody on call")
}
}
func TestEscalationWalksStepsUntilAcked(t *testing.T) {
s, clk, mails := newTestServer(t)
alice, bob := mustUser(t, s, "Alice"), mustUser(t, s, "Bob")
s.st.put("schedule", "sch", &Schedule{ID: "sch", Name: "primary", Members: []string{alice.ID}, RotationHours: 24, Start: clk.Now().Unix()})
s.st.put("policy", "pol", &Policy{ID: "pol", Name: "p", Steps: []Step{
{Schedules: []string{"sch"}},
{DelayMin: 5, Users: []string{bob.ID}},
}})
a, isNew, err := s.RaiseAlert("k1", "test", "db down", "", "critical", "pol")
if err != nil || !isNew {
t.Fatalf("raise: %v new=%v", err, isNew)
}
s.escalate(clk.Now())
if got := s.MessagesSince(userTopic(alice.ID), 0, 10); len(got) != 1 {
t.Fatalf("alice should be paged once, got %d", len(got))
}
if len(s.MessagesSince(userTopic(bob.ID), 0, 10)) != 0 {
t.Fatal("bob must wait for step 2")
}
clk.Advance(4 * time.Minute)
s.escalate(clk.Now())
if len(s.MessagesSince(userTopic(bob.ID), 0, 10)) != 0 {
t.Fatal("step 2 fired early")
}
clk.Advance(2 * time.Minute)
s.escalate(clk.Now())
if len(s.MessagesSince(userTopic(bob.ID), 0, 10)) != 1 {
t.Fatal("bob should be paged after the delay")
}
// A second alert with the same key is deduplicated, not a new page.
b, isNew, _ := s.RaiseAlert("k1", "test", "db down", "still", "critical", "pol")
if isNew || b.ID != a.ID || b.Count != 2 {
t.Fatalf("dedupe: new=%v same=%v count=%d", isNew, b.ID == a.ID, b.Count)
}
// Ack stops escalation entirely.
c, _, _ := s.RaiseAlert("k2", "test", "disk full", "", "warning", "pol")
s.escalate(clk.Now())
if err := s.AckAlert(c.ID, "Alice"); err != nil {
t.Fatal(err)
}
clk.Advance(time.Hour)
before := len(s.MessagesSince(userTopic(bob.ID), 0, 50))
s.escalate(clk.Now())
if len(s.MessagesSince(userTopic(bob.ID), 0, 50)) != before {
t.Fatal("an acked alert kept escalating")
}
// Emails go out on a goroutine; give them a moment.
deadline := time.Now().Add(2 * time.Second)
for len(mails.to()) == 0 && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
if len(mails.to()) == 0 {
t.Fatal("expected email notifications to be sent")
}
}
func TestSnoozeResumesAtSameStep(t *testing.T) {
s, clk, _ := newTestServer(t)
alice := mustUser(t, s, "Alice")
s.st.put("policy", "pol", &Policy{ID: "pol", Name: "p", Steps: []Step{{Users: []string{alice.ID}}}})
a, _, _ := s.RaiseAlert("k", "test", "x", "", "info", "pol")
if err := s.SnoozeAlert(a.ID, "me", 30*time.Minute); err != nil {
t.Fatal(err)
}
s.escalate(clk.Now())
if len(s.MessagesSince(userTopic(alice.ID), 0, 10)) != 0 {
t.Fatal("snoozed alert paged")
}
clk.Advance(31 * time.Minute)
s.escalate(clk.Now())
if len(s.MessagesSince(userTopic(alice.ID), 0, 10)) != 1 {
t.Fatal("alert should page when the snooze ends")
}
}
func TestMonitorHTTPDownAndRecover(t *testing.T) {
s, clk, _ := newTestServer(t)
var mu sync.Mutex
status := 200
backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
w.WriteHeader(status)
io.WriteString(w, "hello world")
}))
defer backend.Close()
m := &Monitor{ID: "m1", Name: "web", Kind: "http", Target: backend.URL, IntervalS: 30, Retries: 2, Enabled: true, AllowPrivate: true, Keyword: "hello"}
if err := validateMonitor(m); err != nil {
t.Fatal(err)
}
s.st.put("monitor", m.ID, m)
step := func() { s.checkMonitor(m, clk.Now()); clk.Advance(30 * time.Second) }
setStatus := func(v int) { mu.Lock(); status = v; mu.Unlock() }
step()
if st := s.monitorState("m1"); st.Status != "up" {
t.Fatalf("want up, got %s (%s)", st.Status, st.Detail)
}
setStatus(500)
step()
if st := s.monitorState("m1"); st.Status != "up" || st.Fails != 1 {
t.Fatalf("one failure must not flip it: %+v", st)
}
step()
if st := s.monitorState("m1"); st.Status != "down" {
t.Fatalf("want down after retries, got %+v", st)
}
if open := s.listAlerts("open", 10); len(open) != 1 || !strings.Contains(open[0].Summary, "web is down") {
t.Fatalf("expected one open alert, got %+v", open)
}
step() // still down: same alert, not a second one
if open := s.listAlerts("open", 10); len(open) != 1 {
t.Fatalf("alert duplicated: %d", len(open))
}
setStatus(200)
step()
if st := s.monitorState("m1"); st.Status != "up" {
t.Fatalf("want recovered, got %+v", st)
}
if open := s.listAlerts("open", 10); len(open) != 0 {
t.Fatalf("alert should have resolved: %+v", open)
}
if up := s.monitorState("m1").Uptime30; up <= 0 || up >= 100 {
t.Fatalf("uptime should reflect the outage, got %v", up)
}
}
func TestMonitorBlocksPrivateTargetsByDefault(t *testing.T) {
s, _, _ := newTestServer(t)
backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
defer backend.Close()
m := &Monitor{Name: "x", Kind: "http", Target: backend.URL}
if _, err := s.runCheck(context.Background(), m); err == nil {
t.Fatal("a loopback target must be refused unless allowPrivate is set")
}
}
func TestHeartbeatMissed(t *testing.T) {
s, clk, _ := newTestServer(t)
m := &Monitor{ID: "hb", Name: "backup", Kind: "heartbeat", IntervalS: 60, GraceS: 10, Enabled: true}
if err := validateMonitor(m); err != nil {
t.Fatal(err)
}
s.st.put("monitor", m.ID, m)
s.checkMonitor(m, clk.Now()) // first look: starts the clock
if s.monitorState("hb").Status != "pending" {
t.Fatal("a heartbeat that was never expected yet should be pending")
}
clk.Advance(30 * time.Second)
if !s.Heartbeat(m.Token) {
t.Fatal("ping rejected")
}
s.checkMonitor(m, clk.Now())
if s.monitorState("hb").Status != "up" {
t.Fatal("want up after a ping")
}
clk.Advance(2 * time.Minute)
for i := 0; i < 2; i++ {
s.checkMonitor(m, clk.Now())
clk.Advance(15 * time.Second)
}
if s.monitorState("hb").Status != "down" {
t.Fatalf("missed heartbeat should be down: %+v", s.monitorState("hb"))
}
if !s.Heartbeat(m.Token) {
t.Fatal("ping rejected")
}
if s.monitorState("hb").Status != "up" || len(s.listAlerts("open", 10)) != 0 {
t.Fatal("a ping should recover a down heartbeat immediately")
}
if s.Heartbeat("nope") {
t.Fatal("unknown token accepted")
}
}
func TestIncidentLifecycle(t *testing.T) {
s, _, _ := newTestServer(t)
u := mustUser(t, s, "Alice")
a, _, _ := s.RaiseAlert("k", "t", "api 500s", "", "critical", "")
in, err := s.DeclareIncident("API outage", "sev1", "errors everywhere", u.ID, u.Name, []string{a.ID})
if err != nil {
t.Fatal(err)
}
if got, _ := s.getAlert(a.ID); got.IncidentID != in.ID {
t.Fatal("alert not linked")
}
st := "identified"
if _, err := s.UpdateIncident(in.ID, u.Name, IncidentPatch{Status: &st, Note: "bad deploy"}); err != nil {
t.Fatal(err)
}
bogus := "exploding"
if _, err := s.UpdateIncident(in.ID, u.Name, IncidentPatch{Status: &bogus}); err == nil {
t.Fatal("invalid status accepted")
}
done := "resolved"
got, _ := s.UpdateIncident(in.ID, u.Name, IncidentPatch{Status: &done})
if got.Resolved == 0 {
t.Fatal("resolved time not set")
}
s.CommentIncident(in.ID, u.Name, "rolled back")
ev := s.st.events("incident", in.ID)
if len(ev) < 4 {
t.Fatalf("timeline too short: %+v", ev)
}
if tpl := s.postmortemTemplate(got); !strings.Contains(tpl, "rolled back") || !strings.Contains(tpl, "API outage") {
t.Fatalf("template missing content:\n%s", tpl)
}
if n := len(s.MessagesSince("incidents", 0, 10)); n != 3 {
t.Fatalf("expected declare + 2 updates on the incidents topic, got %d", n)
}
}
func TestChannelSecretsAreSealedAndRedacted(t *testing.T) {
s, _, _ := newTestServer(t)
ts := httptest.NewServer(s.Handler())
defer ts.Close()
c := newAPIClient(t, ts)
c.setup()
var created map[string]any
c.do("POST", "/api/v1/channels", map[string]any{"name": "slack", "kind": "webhook", "target": "https://hooks.example.com/secret-token", "format": "slack", "enabled": true}, &created)
var raw string
s.st.db.QueryRow(`SELECT body FROM docs WHERE kind='channel'`).Scan(&raw)
if strings.Contains(raw, "secret-token") {
t.Fatal("webhook URL stored in clear text")
}
if created["target"] == "https://hooks.example.com/secret-token" {
t.Fatal("the API returned the secret")
}
// Editing without touching the target keeps the stored secret.
created["name"] = "slack2"
c.do("PUT", "/api/v1/channels/"+created["id"].(string), created, nil)
ch, _ := getOf[Channel](s.st, "channel", created["id"].(string))
if s.st.open(ch.Target) != "https://hooks.example.com/secret-token" || ch.Name != "slack2" {
t.Fatalf("secret lost on edit: %+v", ch)
}
}
// ---------- HTTP API end to end ----------
type apiClient struct {
t *testing.T
ts *httptest.Server
cookie *http.Cookie
}
func newAPIClient(t *testing.T, ts *httptest.Server) *apiClient { return &apiClient{t: t, ts: ts} }
func (c *apiClient) req(method, path string, body any, headers map[string]string) *http.Response {
c.t.Helper()
var rd io.Reader
if body != nil {
b, _ := json.Marshal(body)
rd = strings.NewReader(string(b))
}
r, _ := http.NewRequest(method, c.ts.URL+path, rd)
if body != nil || method != "GET" {
r.Header.Set("Content-Type", "application/json")
}
if c.cookie != nil {
r.AddCookie(c.cookie)
}
for k, v := range headers {
r.Header.Set(k, v)
}
resp, err := http.DefaultClient.Do(r)
if err != nil {
c.t.Fatal(err)
}
return resp
}
func (c *apiClient) do(method, path string, body any, out any) int {
c.t.Helper()
resp := c.req(method, path, body, nil)
defer resp.Body.Close()
if out != nil {
json.NewDecoder(resp.Body).Decode(out)
}
return resp.StatusCode
}
func (c *apiClient) setup() {
c.t.Helper()
resp := c.req("POST", "/api/v1/setup", map[string]string{"name": "Root", "email": "root@example.com", "password": "correct-horse-battery"}, nil)
defer resp.Body.Close()
if resp.StatusCode != 200 {
c.t.Fatalf("setup: %d", resp.StatusCode)
}
for _, ck := range resp.Cookies() {
if ck.Name == sessionCookie {
c.cookie = ck
}
}
if c.cookie == nil {
c.t.Fatal("no session cookie")
}
}
func TestAPIAuthSetupAndTopics(t *testing.T) {
s, _, _ := newTestServer(t)
ts := httptest.NewServer(s.Handler())
defer ts.Close()
c := newAPIClient(t, ts)
if code := c.do("GET", "/api/v1/alerts", nil, nil); code != 401 {
t.Fatalf("unauthenticated API call: %d", code)
}
if code := c.do("POST", "/mytopic", nil, nil); code != 401 {
t.Fatalf("unauthenticated publish: %d", code)
}
c.setup()
if code := c.do("POST", "/api/v1/setup", map[string]string{"name": "x", "email": "x@x.com", "password": "correct-horse-battery"}, nil); code != 403 {
t.Fatalf("second setup must be refused, got %d", code)
}
var key map[string]string
c.do("POST", "/api/v1/apikeys", map[string]string{"name": "ci"}, &key)
if !strings.HasPrefix(key["token"], "sb_") {
t.Fatalf("api key: %v", key)
}
bearer := map[string]string{"Authorization": "Bearer " + key["token"]}
// Subscribe with the API key, then publish ntfy-style with headers.
sub := make(chan string, 4)
go func() {
req, _ := http.NewRequest("GET", ts.URL+"/alerts-room/sse", nil)
req.Header.Set("Authorization", "Bearer "+key["token"])
resp, err := http.DefaultClient.Do(req)
if err != nil {
return
}
defer resp.Body.Close()
sc := bufio.NewScanner(resp.Body)
for sc.Scan() {
if strings.HasPrefix(sc.Text(), "data: ") {
sub <- sc.Text()[6:]
return
}
}
}()
time.Sleep(300 * time.Millisecond)
r, _ := http.NewRequest("POST", ts.URL+"/alerts-room", strings.NewReader("disk is full"))
r.Header.Set("Authorization", "Bearer "+key["token"])
r.Header.Set("Title", "Disk")
r.Header.Set("Priority", "urgent")
r.Header.Set("Tags", "warning,disk")
resp, err := http.DefaultClient.Do(r)
if err != nil || resp.StatusCode != 200 {
t.Fatalf("publish: %v %v", err, resp)
}
select {
case line := <-sub:
var m Message
json.Unmarshal([]byte(line), &m)
if m.Message != "disk is full" || m.Title != "Disk" || m.Priority != 5 || len(m.Tags) != 2 {
t.Fatalf("bad message: %+v", m)
}
case <-time.After(3 * time.Second):
t.Fatal("subscriber got nothing")
}
// The ntfy apps catch up with ?poll=1&since=<last id> and ignore lines without "event".
var ids []string
for _, body := range []string{"one", "two", "three"} {
r, _ := http.NewRequest("POST", ts.URL+"/phone", strings.NewReader(body))
r.Header.Set("Authorization", "Bearer "+key["token"])
resp, err := http.DefaultClient.Do(r)
if err != nil || resp.StatusCode != 200 {
t.Fatalf("publish %s: %v %v", body, err, resp)
}
var pm Message
json.NewDecoder(resp.Body).Decode(&pm)
resp.Body.Close()
if pm.Event != "message" {
t.Fatalf("publish response event: %q", pm.Event)
}
ids = append(ids, pm.ID)
}
pollLines := func(query string) []Message {
req, _ := http.NewRequest("GET", ts.URL+"/phone/json?poll=1&"+query, nil)
req.Header.Set("Authorization", "Bearer "+key["token"])
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
var got []Message
sc := bufio.NewScanner(resp.Body)
for sc.Scan() {
var m Message
json.Unmarshal(sc.Bytes(), &m)
if m.Event != "message" {
t.Fatalf("poll line without event: %s", sc.Text())
}
got = append(got, m)
}
return got
}
if got := pollLines("since=all"); len(got) != 3 {
t.Fatalf("poll since=all: %d messages", len(got))
}
if got := pollLines("since=" + ids[0]); len(got) != 2 || got[0].Message != "two" {
t.Fatalf("poll since=<id>: %+v", got)
}
if got := pollLines("since=" + ids[2]); len(got) != 0 {
t.Fatalf("poll since=<last id>: %+v", got)
}
if got := pollLines("since=10m"); len(got) != 3 {
t.Fatalf("poll since=10m: %d messages", len(got))
}
// Reserved topics are refused.
for _, reserved := range []string{"/api", "/status"} {
r, _ := http.NewRequest("POST", ts.URL+reserved, strings.NewReader("x"))
r.Header.Set("Authorization", "Bearer "+key["token"])
resp, _ := http.DefaultClient.Do(r)
if resp.StatusCode == 200 {
t.Errorf("publish to %s should not succeed", reserved)
}
}
// Inbound alert -> ack -> resolve over the API with the key.
var out map[string]any
c.cookie = nil
if code := c.do("POST", "/api/v1/alerts", map[string]string{"key": "cpu", "summary": "CPU high"}, nil); code != 401 {
t.Fatalf("alert without credentials: %d", code)
}
resp = c.req("POST", "/api/v1/alerts", map[string]string{"key": "cpu", "summary": "CPU high", "severity": "warning"}, bearer)
json.NewDecoder(resp.Body).Decode(&out)
resp.Body.Close()
id, _ := out["id"].(string)
if id == "" {
t.Fatalf("no alert id: %v", out)
}
c.req("POST", "/api/v1/alerts/"+id+"/ack", nil, bearer).Body.Close()
a, _ := s.getAlert(id)
if a.Status != "acked" || a.AckedBy != "Root" {
t.Fatalf("ack: %+v", a)
}
c.req("POST", "/api/v1/alerts", map[string]string{"key": "cpu", "status": "resolved"}, bearer).Body.Close()
if a, _ = s.getAlert(id); a.Status != "resolved" {
t.Fatalf("resolve via key: %+v", a)
}
}
func TestAlertmanagerWebhook(t *testing.T) {
s, _, _ := newTestServer(t)
ts := httptest.NewServer(s.Handler())
defer ts.Close()
c := newAPIClient(t, ts)
c.setup()
payload := map[string]any{"alerts": []map[string]any{{"status": "firing", "fingerprint": "abc",
"labels": map[string]string{"alertname": "HighLatency", "severity": "critical"}, "annotations": map[string]string{"summary": "p99 > 2s"}}}}
c.do("POST", "/api/v1/integrations/alertmanager", payload, nil)
open := s.listAlerts("open", 10)
if len(open) != 1 || open[0].Summary != "p99 > 2s" || open[0].Severity != "critical" {
t.Fatalf("got %+v", open)
}
payload["alerts"].([]map[string]any)[0]["status"] = "resolved"
c.do("POST", "/api/v1/integrations/alertmanager", payload, nil)
if len(s.listAlerts("open", 10)) != 0 {
t.Fatal("not resolved")
}
}
func TestStatusPageOnlyShowsPublic(t *testing.T) {
s, clk, _ := newTestServer(t)
s.st.put("monitor", "a", &Monitor{ID: "a", Name: "Public site", Kind: "http", Target: "http://x", Enabled: true, Public: true})
s.st.put("monitor", "b", &Monitor{ID: "b", Name: "Internal DB", Kind: "tcp", Target: "x:1", Enabled: true})
s.saveState(MonitorState{MonitorID: "a", Status: "up", Since: clk.Now().Unix()})
s.st.put("incident", "i1", &Incident{ID: "i1", Title: "Public incident", Severity: "sev2", Status: "investigating", Public: true, Created: clk.Now().Unix()})
s.st.put("incident", "i2", &Incident{ID: "i2", Title: "Private incident", Severity: "sev1", Status: "investigating", Created: clk.Now().Unix()})
ts := httptest.NewServer(s.Handler())
defer ts.Close()
resp, _ := http.Get(ts.URL + "/status.json")
b, _ := io.ReadAll(resp.Body)
body := string(b)
if !strings.Contains(body, "Public site") || strings.Contains(body, "Internal DB") {
t.Fatalf("monitor visibility wrong: %s", body)
}
if !strings.Contains(body, "Public incident") || strings.Contains(body, "Private incident") {
t.Fatalf("incident visibility wrong: %s", body)
}
if !strings.Contains(body, `"overall":"degraded"`) {
t.Fatalf("overall: %s", body)
}
}
func TestLoginThrottle(t *testing.T) {
s, _, _ := newTestServer(t)
mustUser(t, s, "Alice")
ts := httptest.NewServer(s.Handler())
defer ts.Close()
c := newAPIClient(t, ts)
var last int
for i := 0; i < 12; i++ {
last = c.do("POST", "/api/v1/login", map[string]string{"email": "alice@example.com", "password": "wrong-password-x"}, nil)
}
if last != 429 {
t.Fatalf("expected throttling after repeated failures, got %d", last)
}
// Even the right password is refused while throttled.
if code := c.do("POST", "/api/v1/login", map[string]string{"email": "alice@example.com", "password": "correct-horse-battery"}, nil); code != 429 {
t.Fatalf("throttle should apply to the right password too, got %d", code)
}
}