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>

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=<unix seconds>|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) {

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"`

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=<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"))