feat(ntfy): add the HTTP client, stream manager, and account discovery

Speak the ntfy JSON API directly: subscribe, publish, group topics that share a server and credentials, reconnect with backoff and a since cursor, and list account subscriptions, reservations, and ACL grants.

internal/ntfy/account.go

@@ -0,0 +1,151 @@
+package ntfy
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+
+	"lazyntfy/internal/config"
+)
+
+type accountBody struct {
+	SyncTopic     string `json:"sync_topic"`
+	Subscriptions []struct {
+		BaseURL     string  `json:"base_url"`
+		Topic       string  `json:"topic"`
+		DisplayName *string `json:"display_name"`
+	} `json:"subscriptions"`
+	Reservations []struct {
+		Topic string `json:"topic"`
+	} `json:"reservations"`
+}
+
+type grantBody struct {
+	Topic      string `json:"topic"`
+	Permission string `json:"permission"`
+}
+
+type userBody struct {
+	Grants []grantBody `json:"grants"`
+}
+
+// AccountTopics lists topics visible to this login: account subscriptions,
+// reserved topics, and ACL grants when the token can read /v1/users.
+// Wildcard patterns are skipped. Credentials are not included in errors.
+func (c *Client) AccountTopics(ctx context.Context, host string, auth config.Auth) ([]config.Sub, error) {
+	host = strings.TrimRight(strings.TrimSpace(host), "/")
+	if host == "" || (auth.Token == "" && auth.User == "") {
+		return nil, nil
+	}
+	body, status, err := c.get(ctx, host+"/v1/account", auth)
+	if err != nil {
+		return nil, err
+	}
+	if status == http.StatusUnauthorized || status == http.StatusForbidden {
+		return nil, fmt.Errorf("HTTP %d: not authorized to list topics", status)
+	}
+	if status < 200 || status >= 300 {
+		return nil, fmt.Errorf("HTTP %d from /v1/account", status)
+	}
+	var account accountBody
+	if err := json.Unmarshal(body, &account); err != nil {
+		return nil, fmt.Errorf("account response: %w", err)
+	}
+
+	var subs []config.Sub
+	seen := map[string]bool{}
+	add := func(base, topic, display string) {
+		topic = strings.TrimSpace(topic)
+		if topic == "" || strings.Contains(topic, "*") || topic == account.SyncTopic {
+			return
+		}
+		originHost := host
+		if strings.TrimSpace(base) != "" {
+			originHost = strings.TrimRight(base, "/")
+		}
+		origin, name, err := config.Resolve(topic, originHost)
+		if err != nil {
+			return
+		}
+		key := origin + "\n" + name
+		if seen[key] {
+			return
+		}
+		seen[key] = true
+		subs = append(subs, config.Sub{
+			Origin:  origin,
+			Name:    name,
+			Display: strings.TrimSpace(display),
+			Auth:    auth,
+		})
+	}
+
+	for _, sub := range account.Subscriptions {
+		display := ""
+		if sub.DisplayName != nil {
+			display = *sub.DisplayName
+		}
+		add(sub.BaseURL, sub.Topic, display)
+	}
+	for _, res := range account.Reservations {
+		add("", res.Topic, "")
+	}
+
+	usersBody, usersStatus, err := c.get(ctx, host+"/v1/users", auth)
+	if err != nil || usersStatus < 200 || usersStatus >= 300 {
+		return subs, nil
+	}
+	for _, grant := range parseGrants(usersBody) {
+		perm := strings.ToLower(grant.Permission)
+		if perm == "deny-all" || perm == "deny" || perm == "none" {
+			continue
+		}
+		add("", grant.Topic, "")
+	}
+	return subs, nil
+}
+
+func parseGrants(body []byte) []grantBody {
+	var wrapped struct {
+		Users []userBody `json:"users"`
+	}
+	if err := json.Unmarshal(body, &wrapped); err == nil && wrapped.Users != nil {
+		return flattenGrants(wrapped.Users)
+	}
+	var list []userBody
+	if err := json.Unmarshal(body, &list); err != nil {
+		return nil
+	}
+	return flattenGrants(list)
+}
+
+func flattenGrants(users []userBody) []grantBody {
+	var out []grantBody
+	for _, user := range users {
+		out = append(out, user.Grants...)
+	}
+	return out
+}
+
+func (c *Client) get(ctx context.Context, endpoint string, auth config.Auth) ([]byte, int, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
+	if err != nil {
+		return nil, 0, err
+	}
+	req.Header.Set("User-Agent", userAgent)
+	req.Header.Set("Accept", "application/json")
+	auth.Apply(req)
+	resp, err := c.http().Do(req)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+	if err != nil {
+		return nil, resp.StatusCode, err
+	}
+	return body, resp.StatusCode, nil
+}

internal/ntfy/account_test.go

@@ -0,0 +1,82 @@
+package ntfy
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"lazyntfy/internal/config"
+)
+
+func TestAccountTopics(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Authorization") != "Bearer tk_test" {
+			t.Errorf("missing auth")
+		}
+		switch r.URL.Path {
+		case "/v1/account":
+			_, _ = w.Write([]byte(`{
+				"sync_topic":"st_skip",
+				"subscriptions":[
+					{"base_url":"", "topic":"alerts", "display_name":"Alerts"},
+					{"topic":"st_skip"},
+					{"topic":"ops*"}
+				],
+				"reservations":[{"topic":"backups"}]
+			}`))
+		case "/v1/users":
+			_, _ = w.Write([]byte(`[
+				{"username":"ada","grants":[
+					{"topic":"stats","permission":"read-only"},
+					{"topic":"*","permission":"read-write"},
+					{"topic":"secret","permission":"deny-all"}
+				]}
+			]`))
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	defer srv.Close()
+
+	subs, err := NewClient().AccountTopics(context.Background(), srv.URL, config.Auth{Token: "tk_test"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := map[string]string{}
+	for _, sub := range subs {
+		got[sub.Name] = sub.Display
+		if sub.Auth.Token != "tk_test" {
+			t.Fatal("auth dropped")
+		}
+	}
+	_, hasBackups := got["backups"]
+	_, hasStats := got["stats"]
+	if got["alerts"] != "Alerts" || !hasBackups || !hasStats {
+		t.Fatalf("%v", got)
+	}
+	if _, ok := got["st_skip"]; ok {
+		t.Fatal("sync topic included")
+	}
+	if _, ok := got["secret"]; ok {
+		t.Fatal("denied topic included")
+	}
+	if _, ok := got["*"]; ok || len(got) != 3 {
+		t.Fatalf("%v", got)
+	}
+}
+
+func TestAccountTopicsIgnoresUsersForbidden(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/v1/users" {
+			w.WriteHeader(http.StatusForbidden)
+			return
+		}
+		_, _ = w.Write([]byte(`{"reservations":[{"topic":"alerts"}]}`))
+	}))
+	defer srv.Close()
+	subs, err := NewClient().AccountTopics(context.Background(), srv.URL, config.Auth{Token: "tk_test"})
+	if err != nil || len(subs) != 1 || subs[0].Name != "alerts" {
+		t.Fatalf("%v %+v", err, subs)
+	}
+}

internal/ntfy/client.go

@@ -0,0 +1,178 @@
+package ntfy
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"lazyntfy/internal/config"
+)
+
+const userAgent = "lazyntfy/0.1.0"
+
+// Client talks to an ntfy server over HTTP.
+type Client struct {
+	HTTP *http.Client
+}
+
+// NewClient returns a client whose timeout covers the handshake, not the
+// long-lived response body.
+func NewClient() *Client {
+	return &Client{HTTP: &http.Client{Transport: &http.Transport{
+		Proxy: http.ProxyFromEnvironment,
+		DialContext: (&net.Dialer{
+			Timeout:   10 * time.Second,
+			KeepAlive: 30 * time.Second,
+		}).DialContext,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ResponseHeaderTimeout: 30 * time.Second,
+		IdleConnTimeout:       90 * time.Second,
+		ExpectContinueTimeout: 1 * time.Second,
+	}}}
+}
+
+func (c *Client) http() *http.Client {
+	if c != nil && c.HTTP != nil {
+		return c.HTTP
+	}
+	return http.DefaultClient
+}
+
+// PublishRequest is a plain-text ntfy publish.
+type PublishRequest struct {
+	Title    string
+	Message  string
+	Priority int
+	Tags     []string
+}
+
+// StreamMeta is returned once response headers have been read.
+type StreamMeta struct {
+	Truncated bool
+}
+
+// OpenStream starts a JSON subscription. The caller must close the body.
+func (c *Client) OpenStream(ctx context.Context, endpoint string, auth config.Auth) (*http.Response, StreamMeta, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
+	if err != nil {
+		return nil, StreamMeta{}, err
+	}
+	req.Header.Set("User-Agent", userAgent)
+	req.Header.Set("Accept", "application/x-ndjson")
+	auth.Apply(req)
+	resp, err := c.http().Do(req)
+	if err != nil {
+		return nil, StreamMeta{}, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		defer resp.Body.Close()
+		return nil, StreamMeta{}, httpStatusError(resp)
+	}
+	return resp, StreamMeta{Truncated: resp.Header.Get("X-Messages-Truncated") == "1"}, nil
+}
+
+// Publish sends a message to one topic.
+func (c *Client) Publish(ctx context.Context, origin, name string, auth config.Auth, msg PublishRequest) error {
+	if strings.TrimSpace(msg.Message) == "" {
+		return fmt.Errorf("message is empty")
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, PublishURL(origin, name), strings.NewReader(msg.Message))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("User-Agent", userAgent)
+	req.Header.Set("Content-Type", "text/plain; charset=utf-8")
+	if msg.Title != "" {
+		req.Header.Set("Title", msg.Title)
+	}
+	if msg.Priority > 0 {
+		req.Header.Set("Priority", priorityHeader(msg.Priority))
+	}
+	if len(msg.Tags) > 0 {
+		req.Header.Set("Tags", strings.Join(msg.Tags, ","))
+	}
+	auth.Apply(req)
+	resp, err := c.http().Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return httpStatusError(resp)
+	}
+	_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096))
+	return nil
+}
+
+func httpStatusError(resp *http.Response) error {
+	if resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden {
+		return fmt.Errorf("HTTP %d: not authorized for this topic", resp.StatusCode)
+	}
+	body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+	msg := strings.TrimSpace(string(body))
+	msg = strings.ReplaceAll(msg, "\n", " ")
+	if msg == "" {
+		return fmt.Errorf("HTTP %s", resp.Status)
+	}
+	return fmt.Errorf("HTTP %s: %s", resp.Status, msg)
+}
+
+func priorityHeader(p int) string {
+	switch p {
+	case 1:
+		return "min"
+	case 2:
+		return "low"
+	case 3:
+		return "default"
+	case 4:
+		return "high"
+	case 5:
+		return "max"
+	default:
+		return strconv.Itoa(p)
+	}
+}
+
+// ParsePriority accepts 1-5 or the ntfy names min, low, default, high, and max.
+func ParsePriority(s string) (int, error) {
+	s = strings.ToLower(strings.TrimSpace(s))
+	if s == "" {
+		return 3, nil
+	}
+	switch s {
+	case "min", "1":
+		return 1, nil
+	case "low", "2":
+		return 2, nil
+	case "default", "def", "3":
+		return 3, nil
+	case "high", "4":
+		return 4, nil
+	case "max", "urgent", "5":
+		return 5, nil
+	default:
+		return 0, fmt.Errorf("priority must be 1-5 or min/low/default/high/max")
+	}
+}
+
+// ParseTags splits a comma-separated tag list.
+func ParseTags(s string) []string {
+	if strings.TrimSpace(s) == "" {
+		return nil
+	}
+	parts := strings.Split(s, ",")
+	out := make([]string, 0, len(parts))
+	for _, p := range parts {
+		p = strings.TrimSpace(p)
+		if p != "" {
+			out = append(out, p)
+		}
+	}
+	return out
+}

internal/ntfy/client_test.go

@@ -0,0 +1,102 @@
+package ntfy
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"lazyntfy/internal/config"
+)
+
+func TestOpenStreamAndPublish(t *testing.T) {
+	var published string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Authorization") != "Bearer tk_test" {
+			t.Errorf("auth header missing")
+		}
+		switch r.Method {
+		case http.MethodGet:
+			if !strings.Contains(r.URL.Path, "/alerts/json") {
+				t.Errorf("path %s", r.URL.Path)
+			}
+			w.Header().Set("X-Messages-Truncated", "1")
+			_, _ = io.WriteString(w, "{\"id\":\"m1\",\"time\":10,\"event\":\"message\",\"topic\":\"alerts\",\"message\":\"hi\"}\n")
+		case http.MethodPost:
+			if r.URL.Path != "/alerts" {
+				t.Errorf("post path %s", r.URL.Path)
+			}
+			if r.Header.Get("Title") != "Disk" || r.Header.Get("Priority") != "high" || r.Header.Get("Tags") != "a,b" {
+				t.Errorf("headers %v", r.Header)
+			}
+			b, _ := io.ReadAll(r.Body)
+			published = string(b)
+			w.WriteHeader(http.StatusOK)
+		default:
+			t.Errorf("method %s", r.Method)
+		}
+	}))
+	defer srv.Close()
+
+	c := NewClient()
+	auth := config.Auth{Token: "tk_test"}
+	resp, meta, err := c.OpenStream(context.Background(), srv.URL+"/alerts/json?since=all", auth)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if !meta.Truncated {
+		t.Fatal("expected truncation header")
+	}
+	var n int
+	if err := ReadLines(resp.Body, func(ev Event) error {
+		n++
+		if ev.Message != "hi" {
+			t.Fatalf("%+v", ev)
+		}
+		return nil
+	}, nil); err != nil {
+		t.Fatal(err)
+	}
+	if n != 1 {
+		t.Fatalf("events %d", n)
+	}
+
+	err = c.Publish(context.Background(), srv.URL, "alerts", auth, PublishRequest{
+		Title: "Disk", Message: "full", Priority: 4, Tags: []string{"a", "b"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if published != "full" {
+		t.Fatalf("body %q", published)
+	}
+}
+
+func TestUnauthorizedHidesBody(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusUnauthorized)
+		_, _ = io.WriteString(w, "token tk_supersecret rejected")
+	}))
+	defer srv.Close()
+	_, _, err := NewClient().OpenStream(context.Background(), srv.URL+"/alerts/json", config.Auth{Token: "tk_test"})
+	if err == nil || strings.Contains(err.Error(), "tk_supersecret") || !strings.Contains(err.Error(), "not authorized") {
+		t.Fatalf("%v", err)
+	}
+}
+
+func TestParsePriority(t *testing.T) {
+	p, err := ParsePriority("high")
+	if err != nil || p != 4 {
+		t.Fatal(p, err)
+	}
+	if _, err := ParsePriority("nope"); err == nil {
+		t.Fatal("expected error")
+	}
+	tags := ParseTags(" a, b ,,c ")
+	if strings.Join(tags, "|") != "a|b|c" {
+		t.Fatal(tags)
+	}
+}

internal/ntfy/event.go

@@ -0,0 +1,72 @@
+package ntfy
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+)
+
+// Event is one NDJSON object from GET /topic/json.
+type Event struct {
+	ID         string          `json:"id"`
+	Time       int64           `json:"time"`
+	Expires    int64           `json:"expires,omitempty"`
+	Event      string          `json:"event"`
+	Topic      string          `json:"topic"`
+	Message    string          `json:"message"`
+	Title      string          `json:"title"`
+	Tags       []string        `json:"tags"`
+	Priority   int             `json:"priority"`
+	Click      string          `json:"click"`
+	SequenceID string          `json:"sequence_id,omitempty"`
+	Attachment *Attachment     `json:"attachment,omitempty"`
+	Actions    json.RawMessage `json:"actions,omitempty"`
+}
+
+// Attachment describes a file linked from a message.
+type Attachment struct {
+	Name    string `json:"name"`
+	URL     string `json:"url"`
+	Type    string `json:"type,omitempty"`
+	Size    int64  `json:"size,omitempty"`
+	Expires int64  `json:"expires,omitempty"`
+}
+
+// ParseLine decodes one NDJSON line. ok is false for blank lines.
+func ParseLine(line []byte) (Event, bool, error) {
+	line = bytes.TrimSpace(line)
+	if len(line) == 0 {
+		return Event{}, false, nil
+	}
+	var ev Event
+	if err := json.Unmarshal(line, &ev); err != nil {
+		return Event{}, false, err
+	}
+	if ev.Event == "" {
+		return Event{}, false, fmt.Errorf("missing event")
+	}
+	return ev, true, nil
+}
+
+// ReadLines yields each decoded event until EOF or a read error.
+// Malformed JSON is reported and skipped so one bad line does not kill the stream.
+func ReadLines(r io.Reader, yield func(Event) error, onBad func(error)) error {
+	sc := newScanner(r)
+	for sc.Scan() {
+		ev, ok, err := ParseLine(sc.Bytes())
+		if err != nil {
+			if onBad != nil {
+				onBad(err)
+			}
+			continue
+		}
+		if !ok {
+			continue
+		}
+		if err := yield(ev); err != nil {
+			return err
+		}
+	}
+	return sc.Err()
+}

internal/ntfy/event_test.go

@@ -0,0 +1,37 @@
+package ntfy
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestParseLine(t *testing.T) {
+	ev, ok, err := ParseLine([]byte(`{"id":"abc","time":10,"event":"message","topic":"alerts","message":"hi","priority":5}`))
+	if err != nil || !ok {
+		t.Fatal(err, ok)
+	}
+	if ev.ID != "abc" || ev.Priority != 5 || ev.Message != "hi" {
+		t.Fatalf("%+v", ev)
+	}
+	if _, ok, err := ParseLine([]byte("  ")); err != nil || ok {
+		t.Fatalf("blank: %v %v", err, ok)
+	}
+	if _, _, err := ParseLine([]byte("{")); err == nil {
+		t.Fatal("expected json error")
+	}
+}
+
+func TestReadLinesSkipsGarbage(t *testing.T) {
+	var got []string
+	var bad int
+	err := ReadLines(strings.NewReader("nope\n{\"id\":\"a\",\"time\":1,\"event\":\"open\",\"topic\":\"t\"}\n"), func(ev Event) error {
+		got = append(got, ev.Event)
+		return nil
+	}, func(error) { bad++ })
+	if err != nil {
+		t.Fatal(err)
+	}
+	if bad != 1 || len(got) != 1 || got[0] != "open" {
+		t.Fatalf("bad=%d got=%v", bad, got)
+	}
+}

internal/ntfy/group.go

@@ -0,0 +1,52 @@
+package ntfy
+
+import (
+	"net/url"
+	"strings"
+
+	"lazyntfy/internal/config"
+)
+
+// Group is a set of topics that share an origin and credentials and can
+// therefore share one JSON stream.
+type Group struct {
+	Origin string
+	Names  []string
+	Auth   config.Auth
+}
+
+// URL is the /json subscription endpoint for this group.
+func (g Group) URL(since string) string {
+	parts := make([]string, len(g.Names))
+	for i, name := range g.Names {
+		parts[i] = url.PathEscape(name)
+	}
+	endpoint := g.Origin + "/" + strings.Join(parts, ",") + "/json"
+	if since == "" {
+		return endpoint
+	}
+	q := url.Values{}
+	q.Set("since", since)
+	return endpoint + "?" + q.Encode()
+}
+
+// PublishURL is the POST endpoint for one topic.
+func PublishURL(origin, name string) string {
+	return origin + "/" + url.PathEscape(name)
+}
+
+// GroupSubs buckets subscriptions that can share a connection.
+func GroupSubs(subs []config.Sub) []Group {
+	var out []Group
+	index := map[string]int{}
+	for _, sub := range subs {
+		key := sub.Origin + "\n" + sub.Auth.Token + "\n" + sub.Auth.User + "\n" + sub.Auth.Password
+		if i, ok := index[key]; ok {
+			out[i].Names = append(out[i].Names, sub.Name)
+			continue
+		}
+		index[key] = len(out)
+		out = append(out, Group{Origin: sub.Origin, Names: []string{sub.Name}, Auth: sub.Auth})
+	}
+	return out
+}

internal/ntfy/group_test.go

@@ -0,0 +1,22 @@
+package ntfy
+
+import (
+	"testing"
+
+	"lazyntfy/internal/config"
+)
+
+func TestGroupSubs(t *testing.T) {
+	subs := []config.Sub{
+		{Origin: "https://h", Name: "a", Auth: config.Auth{Token: "t"}},
+		{Origin: "https://h", Name: "b", Auth: config.Auth{Token: "t"}},
+		{Origin: "https://h", Name: "c", Auth: config.Auth{Token: "other"}},
+	}
+	groups := GroupSubs(subs)
+	if len(groups) != 2 || len(groups[0].Names) != 2 {
+		t.Fatalf("%+v", groups)
+	}
+	if groups[0].URL("all") != "https://h/a,b/json?since=all" {
+		t.Fatalf("url %s", groups[0].URL("all"))
+	}
+}

internal/ntfy/manager.go

@@ -0,0 +1,224 @@
+package ntfy
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"lazyntfy/internal/config"
+)
+
+// Notice is a stream lifecycle update or a decoded event.
+type Notice struct {
+	Event  *Event
+	Topics []string
+	Status string
+	Up     bool
+	UpSet  bool
+}
+
+// Manager keeps one reconnecting JSON stream per topic group.
+type Manager struct {
+	client     *Client
+	out        chan Notice
+	reload     chan struct{}
+	mu         sync.Mutex
+	ctx        context.Context
+	subs       []config.Sub
+	since      string
+	cursors    map[string]int64
+	backoff    time.Duration
+	maxBackoff time.Duration
+}
+
+// NewManager starts with no subscriptions until Start is called.
+func NewManager(client *Client) *Manager {
+	if client == nil {
+		client = NewClient()
+	}
+	return &Manager{
+		client:     client,
+		out:        make(chan Notice, 256),
+		reload:     make(chan struct{}, 1),
+		cursors:    map[string]int64{},
+		backoff:    time.Second,
+		maxBackoff: 30 * time.Second,
+	}
+}
+
+// SetBackoff overrides reconnect delays. Tests use a short interval.
+func (m *Manager) SetBackoff(initial, max time.Duration) {
+	m.backoff = initial
+	m.maxBackoff = max
+}
+
+// Prime records already-seen newest times per origin/topic so the first stream
+// does not replay history the client restored from disk. Keys use the same
+// origin + "/" + topic format the manager tracks internally.
+func (m *Manager) Prime(cursors map[string]int64) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for key, ts := range cursors {
+		if ts > m.cursors[key] {
+			m.cursors[key] = ts
+		}
+	}
+}
+
+// Notices is the stream of events and status lines. It is never closed.
+func (m *Manager) Notices() <-chan Notice { return m.out }
+
+// Start subscribes until ctx is cancelled. Call it once.
+func (m *Manager) Start(ctx context.Context, subs []config.Sub, since string) {
+	m.mu.Lock()
+	m.ctx = ctx
+	m.subs = append([]config.Sub(nil), subs...)
+	m.since = since
+	m.mu.Unlock()
+	go m.run()
+}
+
+// Reload drops the current connections and subscribes again.
+func (m *Manager) Reload() {
+	select {
+	case m.reload <- struct{}{}:
+	default:
+	}
+}
+
+// Add appends a topic for this session and reconnects.
+func (m *Manager) Add(sub config.Sub) {
+	m.mu.Lock()
+	m.subs = append(m.subs, sub)
+	m.mu.Unlock()
+	m.Reload()
+}
+
+func (m *Manager) run() {
+	ctx := m.ctx
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		m.mu.Lock()
+		subs := append([]config.Sub(nil), m.subs...)
+		m.mu.Unlock()
+		child, cancel := context.WithCancel(ctx)
+		var wg sync.WaitGroup
+		if len(subs) == 0 {
+			m.emit(child, Notice{Status: "no topics"})
+		}
+		for _, group := range GroupSubs(subs) {
+			wg.Add(1)
+			go func(g Group) {
+				defer wg.Done()
+				m.supervise(child, g)
+			}(group)
+		}
+		stopped := make(chan struct{})
+		go func() {
+			wg.Wait()
+			close(stopped)
+		}()
+		select {
+		case <-ctx.Done():
+			cancel()
+			<-stopped
+			return
+		case <-m.reload:
+			cancel()
+			<-stopped
+		}
+	}
+}
+
+func (m *Manager) supervise(ctx context.Context, g Group) {
+	delay := m.backoff
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		err := m.consume(ctx, g)
+		if ctx.Err() != nil || errors.Is(err, context.Canceled) {
+			return
+		}
+		status := "disconnected, retrying"
+		if err != nil {
+			status = err.Error()
+		}
+		m.emit(ctx, Notice{Topics: g.Names, Status: status, Up: false, UpSet: true})
+		timer := time.NewTimer(delay)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return
+		case <-timer.C:
+		}
+		if delay < m.maxBackoff {
+			delay *= 2
+			if delay > m.maxBackoff {
+				delay = m.maxBackoff
+			}
+		}
+	}
+}
+
+func (m *Manager) consume(ctx context.Context, g Group) error {
+	resp, meta, err := m.client.OpenStream(ctx, g.URL(m.sinceFor(g)), g.Auth)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	m.emit(ctx, Notice{Topics: g.Names, Status: "connected", Up: true, UpSet: true})
+	if meta.Truncated {
+		m.emit(ctx, Notice{Topics: g.Names, Status: "cache truncated to the newest messages", Up: true, UpSet: true})
+	}
+	return ReadLines(resp.Body, func(ev Event) error {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		m.note(g.Origin, ev)
+		copyEv := ev
+		m.emit(ctx, Notice{Event: &copyEv, Topics: []string{ev.Topic}})
+		return nil
+	}, func(err error) {
+		m.emit(ctx, Notice{Status: "skipped malformed event"})
+	})
+}
+
+func (m *Manager) sinceFor(g Group) string {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var max int64
+	for _, name := range g.Names {
+		if ts := m.cursors[g.Origin+"/"+name]; ts > max {
+			max = ts
+		}
+	}
+	if max == 0 {
+		return m.since
+	}
+	return strconv.FormatInt(max, 10)
+}
+
+func (m *Manager) note(origin string, ev Event) {
+	if ev.Time == 0 || ev.Topic == "" || strings.Contains(ev.Topic, ",") {
+		return
+	}
+	key := origin + "/" + ev.Topic
+	m.mu.Lock()
+	if ev.Time > m.cursors[key] {
+		m.cursors[key] = ev.Time
+	}
+	m.mu.Unlock()
+}
+
+func (m *Manager) emit(ctx context.Context, n Notice) {
+	select {
+	case <-ctx.Done():
+	case m.out <- n:
+	}
+}

internal/ntfy/manager_test.go

@@ -0,0 +1,101 @@
+package ntfy
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"lazyntfy/internal/config"
+)
+
+func TestPrimeAvoidsReplayingRestoredHistory(t *testing.T) {
+	gotSince := make(chan string, 1)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotSince <- r.URL.Query().Get("since")
+		w.WriteHeader(http.StatusOK)
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+		<-r.Context().Done()
+	}))
+	defer srv.Close()
+
+	mgr := NewManager(NewClient())
+	mgr.Prime(map[string]int64{srv.URL + "/alerts": 1234})
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	mgr.Start(ctx, []config.Sub{{
+		Origin: srv.URL,
+		Name:   "alerts",
+		Auth:   config.Auth{Token: "tk_test"},
+	}}, "all")
+
+	select {
+	case since := <-gotSince:
+		if since != "1234" {
+			t.Fatalf("first since %q", since)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("timed out waiting for the first request")
+	}
+}
+
+func TestManagerReconnectsWithCursor(t *testing.T) {
+	var hits atomic.Int32
+	secondSince := make(chan string, 1)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		n := hits.Add(1)
+		if r.Header.Get("Authorization") != "Bearer tk_test" {
+			t.Errorf("missing auth")
+		}
+		if n == 1 {
+			if r.URL.Query().Get("since") != "all" {
+				t.Errorf("first since %q", r.URL.Query().Get("since"))
+			}
+			_, _ = io.WriteString(w, "{\"id\":\"m1\",\"time\":100,\"event\":\"message\",\"topic\":\"alerts\",\"message\":\"hi\"}\n")
+			return
+		}
+		secondSince <- r.URL.Query().Get("since")
+		w.WriteHeader(http.StatusOK)
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+		<-r.Context().Done()
+	}))
+	defer srv.Close()
+
+	mgr := NewManager(NewClient())
+	mgr.SetBackoff(5*time.Millisecond, 20*time.Millisecond)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	mgr.Start(ctx, []config.Sub{{
+		Origin: srv.URL,
+		Name:   "alerts",
+		Auth:   config.Auth{Token: "tk_test"},
+	}}, "all")
+
+	deadline := time.After(3 * time.Second)
+	sawMessage := false
+	for !sawMessage {
+		select {
+		case n := <-mgr.Notices():
+			if n.Event != nil && n.Event.ID == "m1" {
+				sawMessage = true
+			}
+		case <-deadline:
+			t.Fatal("timed out waiting for the first message")
+		}
+	}
+	select {
+	case since := <-secondSince:
+		if since != "100" {
+			t.Fatalf("reconnect since %q", since)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatalf("timed out waiting for reconnect, hits=%d", hits.Load())
+	}
+}

internal/ntfy/scan.go

@@ -0,0 +1,13 @@
+package ntfy
+
+import (
+	"bufio"
+	"io"
+)
+
+func newScanner(r io.Reader) *bufio.Scanner {
+	sc := bufio.NewScanner(r)
+	buf := make([]byte, 64*1024)
+	sc.Buffer(buf, 10*1024*1024)
+	return sc
+}