diff --git a/README.md b/README.md index 411434f..d706561 100644 --- a/README.md +++ b/README.md @@ -136,6 +136,21 @@ A topic is 1 to 64 letters, digits, `-` or `_`. Publishing and subscribing both API key (create one on the Account page). Clients that use basic auth can send the key as the password. Messages are kept for `SANDBAR_MESSAGE_DAYS` (default 30). +`?since=` also takes unix seconds, a duration such as `10m`, or the id of the last message you have, and +`?poll=1` returns just that replay and closes, which is how the ntfy apps catch up. + +### Phone app (ntfy) + +1. Create an API key on the Account page and copy it. +2. In the ntfy app, open Settings, then Default server, and enter your Sandbar address + (`https://sandbar.example.com`). Or tick "Use another server" when you add each subscription. +3. Add a subscription. For a page when an escalation reaches you, use your personal topic from the Account page + (`u-`). For everything, use `alerts` and `incidents`. +4. Under Manage users, add a user for the server. The username can be anything and the password is the API key. + +The Android app keeps a connection open, so messages arrive at once. The iPhone app depends on ntfy.sh's push +relay, which Sandbar does not use, so there messages arrive only when the app polls. + Two topics are filled for you: `alerts` gets every alert and its recovery, and `incidents` gets every incident update. Each person also has a personal topic (`u-`, shown on the Account page). Subscribe to it in the ntfy app to be paged on your phone when an escalation policy reaches you. diff --git a/internal/sandbar/api.go b/internal/sandbar/api.go index c2f573b..0f8af84 100644 --- a/internal/sandbar/api.go +++ b/internal/sandbar/api.go @@ -518,11 +518,14 @@ func (s *Server) publishHTTP(w http.ResponseWriter, r *http.Request) { bad(w, err.Error()) return } + m.Event = "message" writeJSON(w, 200, m) } // subscribeHTTP streams a topic as server-sent events (/sse) or JSON lines (/json). -// ?since=|all replays stored messages first. +// ?since= replays stored messages first: "all", unix seconds, a duration such as 10m, or the id +// of the last message the client saw. ?poll=1 sends only that replay and then ends, which is +// how the ntfy apps catch up. func (s *Server) subscribeHTTP(w http.ResponseWriter, r *http.Request) { topic := r.PathValue("topic") if !validTopic(topic) { @@ -543,7 +546,14 @@ func (s *Server) subscribeHTTP(w http.ResponseWriter, r *http.Request) { w.Header().Set("Cache-Control", "no-cache") w.Header().Set("X-Accel-Buffering", "no") bw := bufio.NewWriter(w) + event := func(kind string) { + b, _ := json.Marshal(map[string]any{"id": newID(), "time": s.now().Unix(), "event": kind, "topic": topic}) + fmt.Fprintf(bw, "%s\n", b) + bw.Flush() + fl.Flush() + } write := func(m Message) bool { + m.Event = "message" b, _ := json.Marshal(m) var err error if sse { @@ -560,21 +570,22 @@ func (s *Server) subscribeHTTP(w http.ResponseWriter, r *http.Request) { defer unsub() seen := map[string]bool{} if q := r.URL.Query().Get("since"); q != "" { - since := int64(0) - if q != "all" { - since, _ = strconv.ParseInt(q, 10, 64) - } - for _, m := range s.MessagesSince(topic, since, 500) { + for _, m := range s.replaySince(topic, q) { seen[m.ID] = true if !write(m) { return } } } + if r.URL.Query().Get("poll") == "1" { + return + } if sse { fmt.Fprint(bw, ": connected\n\n") bw.Flush() fl.Flush() + } else { + event("open") } keepalive := time.NewTicker(30 * time.Second) defer keepalive.Stop() @@ -594,15 +605,40 @@ func (s *Server) subscribeHTTP(w http.ResponseWriter, r *http.Request) { case <-keepalive.C: if sse { fmt.Fprint(bw, ": keepalive\n\n") + bw.Flush() + fl.Flush() } else { - fmt.Fprint(bw, "\n") + event("keepalive") } - bw.Flush() - fl.Flush() } } } +// replaySince returns the stored messages a subscriber should see first for a ?since= value. +func (s *Server) replaySince(topic, q string) []Message { + if q == "all" { + return s.MessagesSince(topic, 0, 500) + } + if n, err := strconv.ParseInt(q, 10, 64); err == nil { + return s.MessagesSince(topic, n, 500) + } + if d, err := time.ParseDuration(q); err == nil { + return s.MessagesSince(topic, s.now().Add(-d).Unix(), 500) + } + // Otherwise it is the id of the last message the client has: send what came after it. + var at int64 + if s.st.db.QueryRow(`SELECT time FROM messages WHERE topic=? AND id=?`, topic, q).Scan(&at) != nil { + return nil + } + all := s.MessagesSince(topic, at, 500) + for i, m := range all { + if m.ID == q { + return all[i+1:] + } + } + return all +} + // ---------- alerts in ---------- func (s *Server) inboundAlert(w http.ResponseWriter, r *http.Request) { diff --git a/internal/sandbar/notify.go b/internal/sandbar/notify.go index 328c114..a19791a 100644 --- a/internal/sandbar/notify.go +++ b/internal/sandbar/notify.go @@ -25,8 +25,9 @@ import ( type Message struct { ID string `json:"id"` - Topic string `json:"topic"` 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"` diff --git a/internal/sandbar/sandbar_test.go b/internal/sandbar/sandbar_test.go index a58d204..3610980 100644 --- a/internal/sandbar/sandbar_test.go +++ b/internal/sandbar/sandbar_test.go @@ -453,6 +453,56 @@ func TestAPIAuthSetupAndTopics(t *testing.T) { t.Fatal("subscriber got nothing") } + // The ntfy apps catch up with ?poll=1&since= 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=: %+v", got) + } + if got := pollLines("since=" + ids[2]); len(got) != 0 { + t.Fatalf("poll since=: %+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"))