" + html.EscapeString(c.greeting) + "
") @@ -37,12 +56,30 @@ func (a *authService) render(to, toName, tag string, c mailContent) mailer.Messa text.WriteString(p + "\n\n") h.WriteString("" + html.EscapeString(p) + "
") } + if len(c.lines) > 0 { + h.WriteString("` + html.EscapeString(c.actionLabel) + `
`) } - text.WriteString(footer + "\n") - h.WriteString(`` + html.EscapeString(footer) + `
`) + if c.footer != nil { + text.WriteString(c.footer.text + "\n" + c.footer.url + "\n") + h.WriteString(`` + html.EscapeString(c.footer.text) + ` ` + html.EscapeString(c.footer.url) + `
`) + } else { + footer := "You received this because this address was used on " + a.set.baseURL.Host + + ". If that was not you, you can ignore this mail." + text.WriteString(footer + "\n") + h.WriteString(`` + html.EscapeString(footer) + `
`) + } return mailer.Message{To: to, ToName: toName, Subject: "[" + a.set.fromName + "] " + c.subject, HTML: h.String(), Text: text.String(), Tag: tag} } diff --git a/cmd/server/auth_mail_test.go b/cmd/server/auth_mail_test.go index b9e11f1a..1314f77b 100644 --- a/cmd/server/auth_mail_test.go +++ b/cmd/server/auth_mail_test.go @@ -78,3 +78,17 @@ func TestLogStartupShowsAbsoluteUsersDBPath(t *testing.T) { t.Fatalf("log line = %q, want the absolute path %q", buf.String(), want) } } + +func TestRenderLinesAndFooter(t *testing.T) { + a, _ := newTestAuthService(t) + m := a.render("x@example.org", "X", "notify", mailContent{subject: "S", greeting: "Hi", + lines: []mailLine{{text: "n: offline", url: "https://e.org/#/nodes/ab"}}, + footer: &mailLine{text: "Stop these mails:", url: "https://e.org/#/account/unsubscribe?token=t"}}) + if !strings.Contains(m.HTML, ``) || + !strings.Contains(m.Text, "- n: offline\n https://e.org/#/nodes/ab\n") || + !strings.Contains(m.HTML, `Stop these mails: `) || + !strings.Contains(m.Text, "Stop these mails:\nhttps://e.org/#/account/unsubscribe?token=t\n") || + strings.Contains(m.Text, "If that was not you") { + t.Fatalf("rendered = %q / %q", m.HTML, m.Text) + } +} diff --git a/cmd/server/auth_routes.go b/cmd/server/auth_routes.go index 013840b9..af25f490 100644 --- a/cmd/server/auth_routes.go +++ b/cmd/server/auth_routes.go @@ -41,6 +41,9 @@ func (s *Server) registerAuthRoutes(r *mux.Router) { if s.auth.set.proposals.enabled { s.registerProposalRoutes(r) } + if s.auth.notify != nil { + s.registerNotifyRoutes(r) + } // The webhook exists only when a secret is configured. if s.auth.set.webhookSecret != "" { r.HandleFunc("/api/mail/brevo/webhook", s.handleBrevoWebhook).Methods("POST") diff --git a/cmd/server/auth_service.go b/cmd/server/auth_service.go index 2ffa75c4..44de7a88 100644 --- a/cmd/server/auth_service.go +++ b/cmd/server/auth_service.go @@ -32,10 +32,14 @@ type authService struct { approvedMu sync.Mutex approved atomic.Pointer[[]string] + // notify evaluates node notifications; nil unless + // userManagement.notifications.enabled. + notify *notifier + warnIndistinct sync.Once stop chan struct{} stopOnce sync.Once - wg sync.WaitGroup // the janitor + wg sync.WaitGroup // the janitor and the notifier auditWG sync.WaitGroup // in-flight auditAsync writes } @@ -81,6 +85,16 @@ func (s *Server) initUserManagement(measurementDBPath string) error { defer s.auth.wg.Done() s.auth.janitor(time.Hour) }() + if set.notify.enabled { + s.auth.notify = newNotifier(s.auth, serverNotifySource{s: s}, time.Now) + s.auth.wg.Add(1) + go func() { + defer s.auth.wg.Done() + s.auth.notify.loop(s.auth.stop, set.notify.interval) + }() + log.Printf("[notify] node notifications enabled: every %s, at most %d mails per user and %d in total per 24 hours", + set.notify.interval, set.notify.perUserPerDay, set.notify.maxMailsPerDay) + } return nil } diff --git a/cmd/server/notify_eval.go b/cmd/server/notify_eval.go new file mode 100644 index 00000000..f8e46e65 --- /dev/null +++ b/cmd/server/notify_eval.go @@ -0,0 +1,302 @@ +package main + +import ( + "sort" + "strings" + "time" + + "github.com/meshcore-analyzer/users" +) + +// Node notification evaluator (docs/specs/2026-10-07-node-notifications-design.md). +// evaluateNotifications is a pure function: everything it reads is in +// notifyInput, the time included, so it is tested without a clock or a +// database. The notifier (notify_service.go) gathers the input and acts on +// the result. + +// notifyNode is one analyzer node row the evaluator needs. +type notifyNode struct { + Pubkey string // lowercase hex + Name string + Role string // as stored; compared lowercase + LastSeen string // nodes.last_seen, "" when NULL + BatteryMv *int // latest advert telemetry; nil when the node reports none + Foreign bool // nodes.foreign_advert = 1 +} + +// notifyObserver is one active (not soft-deleted) observer. +type notifyObserver struct { + ID string + Name string + LastSeen string // observers.last_seen, "" when NULL +} + +// notifyInput is everything one evaluation reads. +type notifyInput struct { + Now time.Time + Health HealthThresholds // resolved, as Config.GetHealthThresholds returns it + LowMv int // Config.LowBatteryMv + Accounts map[int64]users.User + Prefs []users.NotifyPrefs + Watches []users.NotifyWatch + States []users.NotifyState + Nodes map[string]notifyNode // watched nodes, plus every foreign node when an admin chose foreign.new + Heard map[string]time.Time // newest packet involving the node in the packet store (node page "Last Heard") + Relayed map[string]time.Time // last relay hop, used for repeaters and rooms only (#1598) + Observers []notifyObserver // nil unless an admin chose observer.offline + // IngestStale: the newest packet in the store is old, so a silent node + // and a silent feed look the same. node.offline and observer.offline + // are then not compared (states unchanged, no changes). + IngestStale bool + // IngestResumedAt: when the last stale period ended (zero: none since + // startup). For a node or observer whose stored state is good, evidence + // older than this counts as heard at this time, so nothing goes offline + // until a full silent window has passed after the feed came back. + IngestResumedAt time.Time +} + +// notifyChange is one transition for one user. +type notifyChange struct { + Event string + Subject string // node pubkey or observer id + Name string // display name at evaluation time + To string // users.NotifyGood, users.NotifyBad or users.NotifyTold + BatteryMv *int // node.battery only + At time.Time +} + +// notifyResult: Changes per user, ordered by event then name; States are +// the rows to write (new or changed), disabled users included; Drop are the +// rows to delete (admin-event rows of users who are no longer admin). +type notifyResult struct { + Changes map[int64][]notifyChange + States []users.NotifyState + Drop []users.NotifyKey +} + +const ( + // batteryHysteresisMv: a low battery is good again only at lowMv + 100. + batteryHysteresisMv = 100 + // foreignBaselineSubject marks that an admin's foreign.new baseline + // exists: the first evaluation after opting in stores every current + // foreign node as told without mailing; afterwards a foreign node + // without a row is new. + foreignBaselineSubject = "*" +) + +var notifyEventRank = map[string]int{ + users.NotifyNodeOffline: 0, users.NotifyNodeBattery: 1, users.NotifyForeignNew: 2, users.NotifyObserverOffline: 3, +} + +// notifyInfra reports the roles whose relay activity counts as heard +// (GetHealthMs and roles.js use the same two). +func notifyInfra(role string) bool { + r := strings.ToLower(role) + return r == "repeater" || r == "room" +} + +// nodeLabel is the node's name made safe for a mail line (mailSafeText), +// or the first 8 hex characters of its key. +func nodeLabel(n notifyNode, known bool, pk string) string { + if name := mailSafeText(n.Name); known && name != "" { + return name + } + if len(pk) > 8 { + return pk[:8] + } + return pk +} + +// nodeOnlineState is good when the node was heard within its role's silent +// window, like roles.js getNodeStatus: the latest of last_seen, the packet +// store's last heard and, for repeaters and rooms, the last relay hop. A +// node missing from the analyzer database is bad. While prev is good, the +// latest evidence is never older than in.IngestResumedAt (the outage grace). +func nodeOnlineState(pk string, n notifyNode, known bool, prev string, in *notifyInput) string { + if !known { + return users.NotifyBad + } + last, _ := parseRelayTS(n.LastSeen) + if t, ok := in.Heard[pk]; ok && t.After(last) { + last = t + } + if notifyInfra(n.Role) { + if t, ok := in.Relayed[pk]; ok && t.After(last) { + last = t + } + } + if prev == users.NotifyGood && last.Before(in.IngestResumedAt) { + last = in.IngestResumedAt + } + _, silentMs := in.Health.GetHealthMs(strings.ToLower(n.Role)) + if last.IsZero() || in.Now.Sub(last) >= time.Duration(silentMs)*time.Millisecond { + return users.NotifyBad + } + return users.NotifyGood +} + +// batteryState applies the low-battery threshold with hysteresis. +func batteryState(mv, lowMv int, prev string) string { + if prev == users.NotifyBad { + if mv >= lowMv+batteryHysteresisMv { + return users.NotifyGood + } + return users.NotifyBad + } + if mv < lowMv { + return users.NotifyBad + } + return users.NotifyGood +} + +// observerState: bad when last_seen is older than observerStaleMinutes, +// good again within observerOnlineMinutes. While prev is good, last_seen +// is never older than resumed (the outage grace, zero for none). +func observerState(o notifyObserver, h HealthThresholds, now, resumed time.Time, prev string) string { + t, ok := parseRelayTS(o.LastSeen) + if prev == users.NotifyGood && !resumed.IsZero() && (!ok || t.Before(resumed)) { + t, ok = resumed, true + } + if !ok { + return users.NotifyBad + } + age := now.Sub(t) + if prev == users.NotifyBad { + if age <= time.Duration(h.ObserverOnlineMinutes)*time.Minute { + return users.NotifyGood + } + return users.NotifyBad + } + if age > time.Duration(h.ObserverStaleMinutes)*time.Minute { + return users.NotifyBad + } + return users.NotifyGood +} + +// evaluateNotifications computes every chosen (user, event, subject) state +// and compares it with the stored one. A subject without a stored state is +// stored without a change (a restart or a new watch never mails); a stored +// state that differs is a change. Users missing from Accounts are skipped. +// A user with watches but no preferences row gets the defaults (enabled, +// node events). A disabled user's states are kept current without changes, +// so turning notifications on again never mails what happened while off. +// Admin events are evaluated only for admins; a non-admin's admin-event +// rows are dropped, like opting out, so a re-promotion starts silently. +func evaluateNotifications(in notifyInput) notifyResult { + prev := make(map[users.NotifyKey]string, len(in.States)) + for _, s := range in.States { + prev[s.NotifyKey] = s.State + } + watched := map[int64][]string{} + for _, w := range in.Watches { + watched[w.UserID] = append(watched[w.UserID], w.Pubkey) + } + var foreign []string + for pk, n := range in.Nodes { + if n.Foreign { + foreign = append(foreign, pk) + } + } + sort.Strings(foreign) + + res := notifyResult{Changes: map[int64][]notifyChange{}} + for _, s := range in.States { + if u, ok := in.Accounts[s.UserID]; ok && u.Role != users.RoleAdmin && users.IsAdminNotifyEvent(s.Event) { + res.Drop = append(res.Drop, s.NotifyKey) + } + } + prefs := make([]users.NotifyPrefs, 0, len(in.Prefs)+len(watched)) + hasPrefs := make(map[int64]bool, len(in.Prefs)) + for _, p := range in.Prefs { + prefs = append(prefs, p) + hasPrefs[p.UserID] = true + } + var defaulted []int64 + for uid := range watched { + if !hasPrefs[uid] { + defaulted = append(defaulted, uid) + } + } + sort.Slice(defaulted, func(i, j int) bool { return defaulted[i] < defaulted[j] }) + for _, uid := range defaulted { + prefs = append(prefs, users.DefaultNotifyPrefs(uid)) + } + mailing := false // Enabled of the user being evaluated + store := func(k users.NotifyKey, state string) { + res.States = append(res.States, users.NotifyState{NotifyKey: k, State: state, ChangedAt: in.Now}) + } + compare := func(k users.NotifyKey, name, state string, mv *int) { + old, had := prev[k] + if had && old == state { + return + } + store(k, state) + if had && mailing { + res.Changes[k.UserID] = append(res.Changes[k.UserID], + notifyChange{Event: k.Event, Subject: k.Subject, Name: name, To: state, BatteryMv: mv, At: in.Now}) + } + } + + for _, p := range prefs { + u, ok := in.Accounts[p.UserID] + if !ok { + continue + } + mailing = p.Enabled + for _, pk := range watched[p.UserID] { + n, known := in.Nodes[pk] + name := nodeLabel(n, known, pk) + if p.Has(users.NotifyNodeOffline) && !in.IngestStale { + k := users.NotifyKey{UserID: p.UserID, Event: users.NotifyNodeOffline, Subject: pk} + compare(k, name, nodeOnlineState(pk, n, known, prev[k], &in), nil) + } + if p.Has(users.NotifyNodeBattery) && known && n.BatteryMv != nil { + k := users.NotifyKey{UserID: p.UserID, Event: users.NotifyNodeBattery, Subject: pk} + compare(k, name, batteryState(*n.BatteryMv, in.LowMv, prev[k]), n.BatteryMv) + } + } + if u.Role != users.RoleAdmin { + continue + } + if p.Has(users.NotifyForeignNew) { + base := users.NotifyKey{UserID: p.UserID, Event: users.NotifyForeignNew, Subject: foreignBaselineSubject} + _, baselined := prev[base] + if !baselined { + store(base, users.NotifyGood) + } + for _, pk := range foreign { + k := users.NotifyKey{UserID: p.UserID, Event: users.NotifyForeignNew, Subject: pk} + if _, told := prev[k]; told { + continue + } + store(k, users.NotifyTold) + if baselined && mailing { + res.Changes[p.UserID] = append(res.Changes[p.UserID], notifyChange{Event: users.NotifyForeignNew, Subject: pk, + Name: nodeLabel(in.Nodes[pk], true, pk), To: users.NotifyTold, At: in.Now}) + } + } + } + if p.Has(users.NotifyObserverOffline) && !in.IngestStale { + for _, o := range in.Observers { + k := users.NotifyKey{UserID: p.UserID, Event: users.NotifyObserverOffline, Subject: o.ID} + name := mailSafeText(o.Name) + if name == "" { + name = o.ID + } + compare(k, name, observerState(o, in.Health, in.Now, in.IngestResumedAt, prev[k]), nil) + } + } + } + for _, list := range res.Changes { + sort.SliceStable(list, func(i, j int) bool { + if a, b := notifyEventRank[list[i].Event], notifyEventRank[list[j].Event]; a != b { + return a < b + } + if list[i].Name != list[j].Name { + return list[i].Name < list[j].Name + } + return list[i].Subject < list[j].Subject + }) + } + return res +} diff --git a/cmd/server/notify_eval_test.go b/cmd/server/notify_eval_test.go new file mode 100644 index 00000000..b3704e35 --- /dev/null +++ b/cmd/server/notify_eval_test.go @@ -0,0 +1,552 @@ +package main + +import ( + "fmt" + "strings" + "testing" + "time" + + "github.com/meshcore-analyzer/users" +) + +var evNow = time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) + +const ( + evUser = int64(1) + evAdmin = int64(2) + evAdmin2 = int64(3) +) + +var ( + evPkA = strings.Repeat("a", 64) + evPkB = strings.Repeat("b", 64) + evPkF = strings.Repeat("f", 64) + evPkG = strings.Repeat("e", 64) +) + +func evAgo(d time.Duration) string { return evNow.Add(-d).Format(time.RFC3339) } +func evMv(v int) *int { return &v } + +// evInput: one user watching evPkA with the default node events, two +// admins without prefs, no stored states. +func evInput() notifyInput { + return notifyInput{ + Now: evNow, Health: (&Config{}).GetHealthThresholds(), LowMv: 3300, + Accounts: map[int64]users.User{ + evUser: {ID: evUser, Role: users.RoleUser, Status: users.StatusActive}, + evAdmin: {ID: evAdmin, Role: users.RoleAdmin, Status: users.StatusActive}, + evAdmin2: {ID: evAdmin2, Role: users.RoleAdmin, Status: users.StatusActive}, + }, + Prefs: []users.NotifyPrefs{{UserID: evUser, Enabled: true, Events: []string{users.NotifyNodeOffline, users.NotifyNodeBattery}}}, + Watches: []users.NotifyWatch{{UserID: evUser, Pubkey: evPkA}}, + Nodes: map[string]notifyNode{}, + Heard: map[string]time.Time{}, + Relayed: map[string]time.Time{}, + } +} + +func evStateRow(uid int64, event, subject, state string) users.NotifyState { + return users.NotifyState{NotifyKey: users.NotifyKey{UserID: uid, Event: event, Subject: subject}, State: state, ChangedAt: evNow.Add(-time.Hour)} +} + +func evState(res notifyResult, uid int64, event, subject string) (string, bool) { + for _, s := range res.States { + if s.UserID == uid && s.Event == event && s.Subject == subject { + return s.State, true + } + } + return "", false +} + +func TestEvaluateFirstEvaluationStoresWithoutChanges(t *testing.T) { + in := evInput() + in.Watches = append(in.Watches, users.NotifyWatch{UserID: evUser, Pubkey: evPkB}) + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(48 * time.Hour)} + in.Nodes[evPkB] = notifyNode{Pubkey: evPkB, Name: "Bravo", Role: "companion", LastSeen: evAgo(time.Hour), BatteryMv: evMv(3100)} + res := evaluateNotifications(in) + if len(res.Changes) != 0 { + t.Fatalf("changes on the first evaluation: %+v", res.Changes) + } + for _, c := range []struct{ event, subject, want string }{ + {users.NotifyNodeOffline, evPkA, users.NotifyBad}, + {users.NotifyNodeOffline, evPkB, users.NotifyGood}, + {users.NotifyNodeBattery, evPkB, users.NotifyBad}, + } { + if got, ok := evState(res, evUser, c.event, c.subject); !ok || got != c.want { + t.Errorf("%s %s = %q (stored %v); want %q", c.event, c.subject[:4], got, ok, c.want) + } + } + if _, ok := evState(res, evUser, users.NotifyNodeBattery, evPkA); ok { + t.Error("battery state stored for a node without telemetry") + } + for _, s := range res.States { + if !s.ChangedAt.Equal(evNow) { + t.Errorf("ChangedAt = %v; want the input's Now", s.ChangedAt) + } + } +} + +func TestEvaluateOfflineAndBack(t *testing.T) { + in := evInput() + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(24 * time.Hour)} + in.States = []users.NotifyState{evStateRow(evUser, users.NotifyNodeOffline, evPkA, users.NotifyGood)} + res := evaluateNotifications(in) + ch := res.Changes[evUser] + if len(ch) != 1 || ch[0].Event != users.NotifyNodeOffline || ch[0].Subject != evPkA || ch[0].To != users.NotifyBad || + ch[0].Name != "Alpha" || !ch[0].At.Equal(evNow) { + t.Fatalf("changes = %+v; want one offline change for Alpha", ch) + } + // One minute inside the 24 h companion window: still online, nothing to write. + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(24*time.Hour - time.Minute)} + if res := evaluateNotifications(in); len(res.Changes) != 0 || len(res.States) != 0 { + t.Fatalf("unchanged online node produced %+v", res) + } + // Stored bad, advert 30 h old, but the packet store heard it an hour ago. + in.States = []users.NotifyState{evStateRow(evUser, users.NotifyNodeOffline, evPkA, users.NotifyBad)} + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(30 * time.Hour)} + in.Heard[evPkA] = evNow.Add(-time.Hour) + if ch := evaluateNotifications(in).Changes[evUser]; len(ch) != 1 || ch[0].To != users.NotifyGood { + t.Fatalf("changes = %+v; want one back-online change", ch) + } +} + +func TestEvaluateRelayAwareInfra(t *testing.T) { + cases := []struct { + name, role, lastSeen string + relayed time.Duration + want string + }{ + {"repeater relaying", "repeater", evAgo(80 * time.Hour), time.Hour, users.NotifyGood}, + {"room relaying, role case", "Room", evAgo(80 * time.Hour), time.Hour, users.NotifyGood}, + {"repeater silent exactly 72 h", "repeater", evAgo(72 * time.Hour), 0, users.NotifyBad}, + {"repeater 71 h", "repeater", evAgo(71 * time.Hour), 0, users.NotifyGood}, + {"companion: relay time ignored", "companion", evAgo(30 * time.Hour), time.Hour, users.NotifyBad}, + {"no timestamp at all", "repeater", "", 0, users.NotifyBad}, + } + for _, c := range cases { + in := evInput() + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Role: c.role, LastSeen: c.lastSeen} + if c.relayed > 0 { + in.Relayed[evPkA] = evNow.Add(-c.relayed) + } + if got, _ := evState(evaluateNotifications(in), evUser, users.NotifyNodeOffline, evPkA); got != c.want { + t.Errorf("%s: %q; want %q", c.name, got, c.want) + } + } +} + +func TestEvaluateBatteryHysteresis(t *testing.T) { + cases := []struct { + prev string + mv int + want string + change bool + }{ + {"", 3299, users.NotifyBad, false}, + {"", 3300, users.NotifyGood, false}, + {"", 3350, users.NotifyGood, false}, + {users.NotifyGood, 3299, users.NotifyBad, true}, + {users.NotifyGood, 3300, users.NotifyGood, false}, + {users.NotifyBad, 3399, users.NotifyBad, false}, + {users.NotifyBad, 3400, users.NotifyGood, true}, + } + for _, c := range cases { + in := evInput() + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(time.Hour), BatteryMv: evMv(c.mv)} + in.States = []users.NotifyState{evStateRow(evUser, users.NotifyNodeOffline, evPkA, users.NotifyGood)} + if c.prev != "" { + in.States = append(in.States, evStateRow(evUser, users.NotifyNodeBattery, evPkA, c.prev)) + } + res := evaluateNotifications(in) + got, stored := evState(res, evUser, users.NotifyNodeBattery, evPkA) + if !stored { + got = c.prev + } + ch := res.Changes[evUser] + if got != c.want || (len(ch) == 1) != c.change { + t.Errorf("prev %q, %d mV: state %q, changes %+v; want %q, change %v", c.prev, c.mv, got, ch, c.want, c.change) + continue + } + if c.change && (ch[0].Event != users.NotifyNodeBattery || ch[0].BatteryMv == nil || *ch[0].BatteryMv != c.mv) { + t.Errorf("prev %q, %d mV: change = %+v", c.prev, c.mv, ch[0]) + } + } +} + +func TestEvaluateAbsentNodeIsOfflineOnce(t *testing.T) { + in := evInput() // evPkA is watched but not in the analyzer database + in.States = []users.NotifyState{ + evStateRow(evUser, users.NotifyNodeOffline, evPkA, users.NotifyGood), + evStateRow(evUser, users.NotifyNodeBattery, evPkA, users.NotifyBad), + } + ch := evaluateNotifications(in).Changes[evUser] + if len(ch) != 1 || ch[0].Event != users.NotifyNodeOffline || ch[0].To != users.NotifyBad || ch[0].Name != evPkA[:8] { + t.Fatalf("changes = %+v; want one offline change named by the key prefix", ch) + } + in.States[0] = evStateRow(evUser, users.NotifyNodeOffline, evPkA, users.NotifyBad) + if res := evaluateNotifications(in); len(res.Changes) != 0 || len(res.States) != 0 { + t.Fatalf("absent node evaluated again: %+v", res) + } +} + +func TestEvaluateForeignOncePerAdmin(t *testing.T) { + in := evInput() + in.Prefs[0].Events = append(in.Prefs[0].Events, users.NotifyForeignNew) // a non-admin with an admin event + in.Prefs = append(in.Prefs, users.NotifyPrefs{UserID: evAdmin, Enabled: true, Events: []string{users.NotifyForeignNew}}) + in.Nodes[evPkF] = notifyNode{Pubkey: evPkF, Name: "Foxtrot", Foreign: true, LastSeen: evAgo(time.Hour)} + res := evaluateNotifications(in) + if len(res.Changes[evAdmin]) != 0 { + t.Fatalf("first evaluation mailed: %+v", res.Changes[evAdmin]) + } + if s, ok := evState(res, evAdmin, users.NotifyForeignNew, foreignBaselineSubject); !ok || s != users.NotifyGood { + t.Fatalf("baseline row = %q, %v", s, ok) + } + if s, ok := evState(res, evAdmin, users.NotifyForeignNew, evPkF); !ok || s != users.NotifyTold { + t.Fatalf("existing foreign node = %q, %v; want told", s, ok) + } + if _, ok := evState(res, evUser, users.NotifyForeignNew, evPkF); ok { + t.Fatal("non-admin evaluated for foreign.new") + } + + // Next tick: a second foreign node appears; a second admin opts in now. + in.States = res.States + in.Prefs = append(in.Prefs, users.NotifyPrefs{UserID: evAdmin2, Enabled: true, Events: []string{users.NotifyForeignNew}}) + in.Nodes[evPkG] = notifyNode{Pubkey: evPkG, Name: "Golf", Foreign: true} + res = evaluateNotifications(in) + ch := res.Changes[evAdmin] + if len(ch) != 1 || ch[0].Subject != evPkG || ch[0].To != users.NotifyTold || ch[0].Name != "Golf" { + t.Fatalf("admin changes = %+v; want Golf once", ch) + } + if len(res.Changes[evAdmin2]) != 0 { + t.Fatalf("second admin's first evaluation mailed: %+v", res.Changes[evAdmin2]) + } + + in.States = append(in.States, res.States...) + if res := evaluateNotifications(in); len(res.Changes) != 0 { + t.Fatalf("foreign nodes mailed twice: %+v", res.Changes) + } +} + +func TestEvaluateObserverOfflineAndBack(t *testing.T) { + cases := []struct { + prev string + age time.Duration + want string + change bool + }{ + {"", 2 * time.Hour, users.NotifyGood, false}, + {"", 25 * time.Hour, users.NotifyBad, false}, + {users.NotifyGood, 24 * time.Hour, users.NotifyGood, false}, + {users.NotifyGood, 24*time.Hour + time.Minute, users.NotifyBad, true}, + {users.NotifyBad, 2 * time.Hour, users.NotifyBad, false}, + {users.NotifyBad, 60 * time.Minute, users.NotifyGood, true}, + } + for _, c := range cases { + in := evInput() + in.Prefs = []users.NotifyPrefs{{UserID: evAdmin, Enabled: true, Events: []string{users.NotifyObserverOffline}}} + in.Watches = nil + in.Observers = []notifyObserver{{ID: "OBS1", Name: "Roof", LastSeen: evAgo(c.age)}} + if c.prev != "" { + in.States = []users.NotifyState{evStateRow(evAdmin, users.NotifyObserverOffline, "OBS1", c.prev)} + } + res := evaluateNotifications(in) + got, stored := evState(res, evAdmin, users.NotifyObserverOffline, "OBS1") + if !stored { + got = c.prev + } + ch := res.Changes[evAdmin] + if got != c.want || (len(ch) == 1) != c.change || (c.change && ch[0].Name != "Roof") { + t.Errorf("prev %q, age %v: state %q, changes %+v; want %q, change %v", c.prev, c.age, got, ch, c.want, c.change) + } + } + in := evInput() + in.Prefs = []users.NotifyPrefs{{UserID: evAdmin, Enabled: true, Events: []string{users.NotifyObserverOffline}}} + in.Observers = []notifyObserver{{ID: "OBS2"}} + if s, _ := evState(evaluateNotifications(in), evAdmin, users.NotifyObserverOffline, "OBS2"); s != users.NotifyBad { + t.Fatalf("observer without last_seen = %q; want bad", s) + } +} + +func TestEvaluateOnlyChosenEventsAndKnownAccounts(t *testing.T) { + in := evInput() + in.Prefs[0].Events = []string{users.NotifyNodeBattery, users.NotifyObserverOffline} // observer.offline: the user is no admin + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Role: "companion", LastSeen: evAgo(48 * time.Hour), BatteryMv: evMv(4000)} + in.Observers = []notifyObserver{{ID: "OBS1", LastSeen: evAgo(time.Hour)}} + res := evaluateNotifications(in) + if len(res.States) != 1 || res.States[0].Event != users.NotifyNodeBattery { + t.Fatalf("states = %+v; want only node.battery", res.States) + } + in.Accounts = map[int64]users.User{} // the user was deleted after the read + if res := evaluateNotifications(in); len(res.States) != 0 { + t.Fatalf("deleted user evaluated: %+v", res.States) + } +} + +func TestEvaluateChangesAreOrderedPerUser(t *testing.T) { + in := evInput() + in.Watches = []users.NotifyWatch{{UserID: evUser, Pubkey: evPkB}, {UserID: evUser, Pubkey: evPkA}} + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(48 * time.Hour), BatteryMv: evMv(3000)} + in.Nodes[evPkB] = notifyNode{Pubkey: evPkB, Name: "Bravo", Role: "companion", LastSeen: evAgo(48 * time.Hour)} + in.States = []users.NotifyState{ + evStateRow(evUser, users.NotifyNodeOffline, evPkA, users.NotifyGood), + evStateRow(evUser, users.NotifyNodeOffline, evPkB, users.NotifyGood), + evStateRow(evUser, users.NotifyNodeBattery, evPkA, users.NotifyGood), + } + var got []string + for _, c := range evaluateNotifications(in).Changes[evUser] { + got = append(got, c.Event+" "+c.Name) + } + if want := "node.offline Alpha|node.offline Bravo|node.battery Alpha"; strings.Join(got, "|") != want { + t.Fatalf("order = %q; want %q", strings.Join(got, "|"), want) + } +} + +// 100 users x 50 watches over 2,000 nodes: the spec's Performance figure +// (5,000 state rows per event type). Run with -bench and quote the result +// in the PR. +func BenchmarkEvaluateNotifications(b *testing.B) { + in := notifyInput{Now: evNow, Health: (&Config{}).GetHealthThresholds(), LowMv: 3300, + Accounts: map[int64]users.User{}, Nodes: map[string]notifyNode{}, Heard: map[string]time.Time{}, Relayed: map[string]time.Time{}} + for i := 0; i < 2000; i++ { + pk := fmt.Sprintf("%064x", i) + in.Nodes[pk] = notifyNode{Pubkey: pk, Role: "repeater", LastSeen: evAgo(time.Duration(i) * time.Minute), BatteryMv: evMv(3200 + i%300)} + } + for u := int64(1); u <= 100; u++ { + in.Accounts[u] = users.User{ID: u, Role: users.RoleUser, Status: users.StatusActive} + in.Prefs = append(in.Prefs, users.NotifyPrefs{UserID: u, Enabled: true, Events: users.NodeNotifyEvents}) + for w := 0; w < 50; w++ { + in.Watches = append(in.Watches, users.NotifyWatch{UserID: u, Pubkey: fmt.Sprintf("%064x", (int(u)*37+w*11)%2000)}) + } + } + in.States = evaluateNotifications(in).States + b.ResetTimer() + for i := 0; i < b.N; i++ { + evaluateNotifications(in) + } +} + +// P7: an admin who loses the admin role loses their admin-event states, like +// opting out of those events, so a re-promotion starts a fresh baseline. +func TestEvaluateDemotedAdminDropsAdminEventStates(t *testing.T) { + in := evInput() + in.Prefs = append(in.Prefs, users.NotifyPrefs{UserID: evAdmin, Enabled: true, + Events: []string{users.NotifyNodeOffline, users.NotifyForeignNew, users.NotifyObserverOffline}}) + in.Watches = append(in.Watches, users.NotifyWatch{UserID: evAdmin, Pubkey: evPkA}) + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(time.Hour)} + in.Nodes[evPkF] = notifyNode{Pubkey: evPkF, Name: "Foxtrot", Foreign: true} + in.Observers = []notifyObserver{{ID: "OBS1", Name: "Roof", LastSeen: evAgo(time.Hour)}} + in.States = evaluateNotifications(in).States + if s, ok := evState(notifyResult{States: in.States}, evAdmin, users.NotifyForeignNew, foreignBaselineSubject); !ok || s != users.NotifyGood { + t.Fatalf("baseline row = %q, %v", s, ok) + } + + // Demoted: the admin-event rows are dropped, the node rows stay, nothing mailed. + admin := in.Accounts[evAdmin] + admin.Role = users.RoleUser + in.Accounts[evAdmin] = admin + in.Nodes[evPkG] = notifyNode{Pubkey: evPkG, Name: "Golf", Foreign: true} + in.Observers[0].LastSeen = evAgo(48 * time.Hour) + res := evaluateNotifications(in) + if len(res.Changes) != 0 || len(res.States) != 0 { + t.Fatalf("demoted admin evaluated: %+v", res) + } + want := map[users.NotifyKey]bool{ + {UserID: evAdmin, Event: users.NotifyForeignNew, Subject: foreignBaselineSubject}: true, + {UserID: evAdmin, Event: users.NotifyForeignNew, Subject: evPkF}: true, + {UserID: evAdmin, Event: users.NotifyObserverOffline, Subject: "OBS1"}: true, + } + if len(res.Drop) != len(want) { + t.Fatalf("drop = %+v; want %d admin-event keys", res.Drop, len(want)) + } + for _, k := range res.Drop { + if !want[k] { + t.Errorf("dropped %+v", k) + } + } + dropped := map[users.NotifyKey]bool{} + for _, k := range res.Drop { + dropped[k] = true + } + var kept []users.NotifyState + for _, s := range in.States { + if !dropped[s.NotifyKey] { + kept = append(kept, s) + } + } + in.States = kept + if res := evaluateNotifications(in); len(res.Drop) != 0 { + t.Fatalf("dropped again: %+v", res.Drop) + } + + // Re-promoted: Golf appeared and the observer went offline meanwhile; the + // first evaluation stores a new baseline and mails nothing. + admin.Role = users.RoleAdmin + in.Accounts[evAdmin] = admin + res = evaluateNotifications(in) + if len(res.Changes) != 0 || len(res.Drop) != 0 { + t.Fatalf("re-promoted admin: changes %+v, drop %+v", res.Changes, res.Drop) + } + for _, c := range []struct{ event, subject, want string }{ + {users.NotifyForeignNew, foreignBaselineSubject, users.NotifyGood}, + {users.NotifyForeignNew, evPkG, users.NotifyTold}, + {users.NotifyObserverOffline, "OBS1", users.NotifyBad}, + } { + if got, ok := evState(res, evAdmin, c.event, c.subject); !ok || got != c.want { + t.Errorf("%s %s = %q (stored %v); want %q", c.event, c.subject, got, ok, c.want) + } + } +} + +// A user who watches nodes but has no preferences row (never opened the +// settings, or created before the notification schema) gets the defaults: +// enabled, node events on, admin events off. +func TestEvaluateWatcherWithoutPrefsGetsDefaults(t *testing.T) { + in := evInput() + in.Prefs = nil + in.Watches = []users.NotifyWatch{{UserID: evAdmin, Pubkey: evPkA}} + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(48 * time.Hour), BatteryMv: evMv(3000)} + in.Nodes[evPkF] = notifyNode{Pubkey: evPkF, Name: "Foxtrot", Foreign: true} + in.Observers = []notifyObserver{{ID: "OBS1", LastSeen: evAgo(time.Hour)}} + in.States = []users.NotifyState{ + evStateRow(evAdmin, users.NotifyNodeOffline, evPkA, users.NotifyGood), + evStateRow(evAdmin, users.NotifyNodeBattery, evPkA, users.NotifyGood), + } + res := evaluateNotifications(in) + var got []string + for _, c := range res.Changes[evAdmin] { + got = append(got, c.Event+" "+c.To) + } + if want := "node.offline bad|node.battery bad"; strings.Join(got, "|") != want { + t.Fatalf("changes = %q; want %q", strings.Join(got, "|"), want) + } + for _, s := range res.States { + if users.IsAdminNotifyEvent(s.Event) { + t.Errorf("admin event evaluated without opting in: %+v", s) + } + } + in.Accounts = map[int64]users.User{} + if res := evaluateNotifications(in); len(res.States) != 0 { + t.Fatalf("watcher without an account evaluated: %+v", res.States) + } +} + +// A disabled user keeps being evaluated: states follow the node, nothing is +// mailed, and re-enabling does not mail what changed while off. +func TestEvaluateDisabledUserUpdatesStatesSilently(t *testing.T) { + in := evInput() + in.Prefs[0].Enabled = false + in.Prefs = append(in.Prefs, users.NotifyPrefs{UserID: evAdmin, Enabled: false, Events: []string{users.NotifyForeignNew}}) + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(time.Hour)} + in.Nodes[evPkF] = notifyNode{Pubkey: evPkF, Name: "Foxtrot", Foreign: true} + in.States = evaluateNotifications(in).States + + // While off: Alpha goes offline and Golf appears. + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(48 * time.Hour)} + in.Nodes[evPkG] = notifyNode{Pubkey: evPkG, Name: "Golf", Foreign: true} + res := evaluateNotifications(in) + if len(res.Changes) != 0 { + t.Fatalf("disabled users mailed: %+v", res.Changes) + } + if s, ok := evState(res, evUser, users.NotifyNodeOffline, evPkA); !ok || s != users.NotifyBad { + t.Fatalf("disabled user's offline state = %q, %v; want bad stored", s, ok) + } + if s, ok := evState(res, evAdmin, users.NotifyForeignNew, evPkG); !ok || s != users.NotifyTold { + t.Fatalf("disabled admin's foreign state = %q, %v; want told stored", s, ok) + } + in.States = append(in.States, res.States...) + + // Re-enabled: the next evaluation has nothing to send. + in.Prefs[0].Enabled = true + in.Prefs[1].Enabled = true + if res := evaluateNotifications(in); len(res.Changes) != 0 || len(res.States) != 0 { + t.Fatalf("re-enabling mailed old changes: %+v", res) + } +} + +// While ingest is stale nobody can tell a silent node from a silent +// MQTT feed: offline comparisons pause with their states unchanged; +// battery and foreign nodes go on. +func TestEvaluateIngestStalePausesOfflineChecks(t *testing.T) { + in := evInput() + in.IngestStale = true + in.Watches = append(in.Watches, users.NotifyWatch{UserID: evUser, Pubkey: evPkB}) + in.Prefs = append(in.Prefs, users.NotifyPrefs{UserID: evAdmin, Enabled: true, Events: []string{users.NotifyForeignNew, users.NotifyObserverOffline}}) + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(48 * time.Hour), BatteryMv: evMv(3100)} + in.Nodes[evPkF] = notifyNode{Pubkey: evPkF, Name: "Foxtrot", Foreign: true} + in.Observers = []notifyObserver{{ID: "OBS1", Name: "Roof", LastSeen: evAgo(48 * time.Hour)}} + in.States = []users.NotifyState{ + evStateRow(evUser, users.NotifyNodeOffline, evPkA, users.NotifyGood), + evStateRow(evUser, users.NotifyNodeBattery, evPkA, users.NotifyGood), + evStateRow(evAdmin, users.NotifyObserverOffline, "OBS1", users.NotifyGood), + evStateRow(evAdmin, users.NotifyForeignNew, foreignBaselineSubject, users.NotifyGood), + } + res := evaluateNotifications(in) + for _, k := range []struct { + uid int64 + event, subject string + }{{evUser, users.NotifyNodeOffline, evPkA}, {evUser, users.NotifyNodeOffline, evPkB}, {evAdmin, users.NotifyObserverOffline, "OBS1"}} { + if s, ok := evState(res, k.uid, k.event, k.subject); ok { + t.Errorf("%s %s written as %q while ingest is stale", k.event, k.subject, s) + } + } + if ch := res.Changes[evUser]; len(ch) != 1 || ch[0].Event != users.NotifyNodeBattery || ch[0].To != users.NotifyBad { + t.Fatalf("user changes = %+v; want only the battery", ch) + } + if ch := res.Changes[evAdmin]; len(ch) != 1 || ch[0].Event != users.NotifyForeignNew || ch[0].Subject != evPkF { + t.Fatalf("admin changes = %+v; want only the foreign node", ch) + } + + in.IngestStale = false + res = evaluateNotifications(in) + if s, _ := evState(res, evUser, users.NotifyNodeOffline, evPkA); s != users.NotifyBad { + t.Fatalf("offline check after ingest resumed = %q; want bad", s) + } + if s, _ := evState(res, evAdmin, users.NotifyObserverOffline, "OBS1"); s != users.NotifyBad { + t.Fatalf("observer check after ingest resumed = %q; want bad", s) + } +} + +// After a feed outage, evidence older than the recovery counts as heard at +// the recovery: no node or observer goes offline until a full silent window +// has passed since then. A state that is already bad is not lifted by it. +func TestEvaluateGraceAfterIngestResumes(t *testing.T) { + in := evInput() + in.Prefs = append(in.Prefs, users.NotifyPrefs{UserID: evAdmin, Enabled: true, Events: []string{users.NotifyObserverOffline}}) + in.Watches = append(in.Watches, users.NotifyWatch{UserID: evUser, Pubkey: evPkB}) + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Alpha", Role: "companion", LastSeen: evAgo(25 * time.Hour)} + in.Nodes[evPkB] = notifyNode{Pubkey: evPkB, Name: "Bravo", Role: "companion", LastSeen: evAgo(48 * time.Hour)} + in.Observers = []notifyObserver{{ID: "OBS1", Name: "Roof", LastSeen: evAgo(25 * time.Hour)}, {ID: "OBS2", Name: "Mast", LastSeen: evAgo(48 * time.Hour)}} + in.States = []users.NotifyState{ + evStateRow(evUser, users.NotifyNodeOffline, evPkA, users.NotifyGood), + evStateRow(evUser, users.NotifyNodeOffline, evPkB, users.NotifyBad), + evStateRow(evAdmin, users.NotifyObserverOffline, "OBS1", users.NotifyGood), + evStateRow(evAdmin, users.NotifyObserverOffline, "OBS2", users.NotifyBad), + } + in.IngestResumedAt = evNow.Add(-10 * time.Minute) + if res := evaluateNotifications(in); len(res.Changes) != 0 || len(res.States) != 0 { + t.Fatalf("10 minutes after recovery: changes %+v, states %+v; want none", res.Changes, res.States) + } + + // One silent window (24 h companions, observerStaleMinutes 1440) plus a + // minute after recovery, still no new evidence: offline. + in.Now = in.IngestResumedAt.Add(24*time.Hour + time.Minute) + res := evaluateNotifications(in) + if ch := res.Changes[evUser]; len(ch) != 1 || ch[0].Subject != evPkA || ch[0].To != users.NotifyBad { + t.Fatalf("node changes after the window = %+v; want Alpha offline", ch) + } + if ch := res.Changes[evAdmin]; len(ch) != 1 || ch[0].Subject != "OBS1" || ch[0].To != users.NotifyBad { + t.Fatalf("observer changes after the window = %+v; want Roof offline", ch) + } + + // A node heard after the recovery behaves as without the grace. + in.Now = evNow + in.Nodes[evPkB] = notifyNode{Pubkey: evPkB, Name: "Bravo", Role: "companion", LastSeen: evAgo(5 * time.Minute)} + in.Observers[1].LastSeen = evAgo(5 * time.Minute) + res = evaluateNotifications(in) + if ch := res.Changes[evUser]; len(ch) != 1 || ch[0].Subject != evPkB || ch[0].To != users.NotifyGood { + t.Fatalf("node heard after recovery = %+v; want Bravo back online", ch) + } + if ch := res.Changes[evAdmin]; len(ch) != 1 || ch[0].Subject != "OBS2" || ch[0].To != users.NotifyGood { + t.Fatalf("observer heard after recovery = %+v; want Mast back online", ch) + } +} diff --git a/cmd/server/notify_handlers.go b/cmd/server/notify_handlers.go new file mode 100644 index 00000000..4d0b856c --- /dev/null +++ b/cmd/server/notify_handlers.go @@ -0,0 +1,312 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "log" + "net/http" + "regexp" + "strconv" + "strings" + "time" + + "github.com/gorilla/mux" + "github.com/meshcore-analyzer/users" +) + +// Node notification routes (docs/specs/2026-10-07-node-notifications-design.md). +// Registered only when userManagement.notifications.enabled (see +// registerAuthRoutes). + +type notifyPrefsRequest struct { + Enabled bool `json:"enabled"` + Events []string `json:"events"` +} + +type notifyWatchJSON struct { + Pubkey string `json:"pubkey"` + Name string `json:"name"` // "" when unnamed or no longer in the analyzer database + Known bool `json:"known"` // false once the node left the analyzer database + CreatedAt string `json:"createdAt"` +} + +type notifyLimitsJSON struct { + MaxWatches int `json:"maxWatches"` + PerUserPerDay int `json:"perUserPerDay"` + MailsLast24h int `json:"mailsLast24h"` +} + +// notifyAccountJSON is the caller's notification state: GET answers it, and +// so does every change, so the client can replace its copy. +type notifyAccountJSON struct { + Enabled bool `json:"enabled"` + Events []string `json:"events"` + AvailableEvents []string `json:"availableEvents"` + Watches []notifyWatchJSON `json:"watches"` + Limits notifyLimitsJSON `json:"limits"` +} + +type watchMyNodesJSON struct { + Added int `json:"added"` + Already int `json:"already"` + Skipped int `json:"skipped"` // malformed, not in the analyzer database, or over the limit + Account notifyAccountJSON `json:"account"` +} + +func (s *Server) registerNotifyRoutes(r *mux.Router) { + r.HandleFunc("/api/account/notifications", s.withUser(s.handleNotifyGet)).Methods("GET") + r.HandleFunc("/api/account/notifications", s.withUser(s.handleNotifyPut)).Methods("PUT") + r.HandleFunc("/api/account/notifications/watches/{pubkey}", s.withUser(s.handleWatchPut)).Methods("PUT") + r.HandleFunc("/api/account/notifications/watches/{pubkey}", s.withUser(s.handleWatchDelete)).Methods("DELETE") + r.HandleFunc("/api/account/notifications/watch-my-nodes", s.withUser(s.handleWatchMyNodes)).Methods("POST") + // No session: the link comes from a mail, and the provider's one-click + // POST (List-Unsubscribe-Post) carries neither a cookie nor our Origin. + r.HandleFunc("/api/notifications/unsubscribe", s.handleUnsubscribeGet).Methods("GET") + r.HandleFunc("/api/notifications/unsubscribe", s.handleUnsubscribePost).Methods("POST") +} + +var notifyPubkeyRE = regexp.MustCompile(`^[0-9a-f]{64}$`) + +// notifyPubkey lowercases raw and reports whether it is a 64-hex-character key. +func notifyPubkey(raw string) (string, bool) { + pk := strings.ToLower(strings.TrimSpace(raw)) + return pk, notifyPubkeyRE.MatchString(pk) +} + +func availableNotifyEvents(u *users.User) []string { + out := append([]string{}, users.NodeNotifyEvents...) + if u.Role == users.RoleAdmin { + out = append(out, users.AdminNotifyEvents...) + } + return out +} + +func notifyInternal(w http.ResponseWriter, op string, uid int64, err error) { + log.Printf("[notify] %s for user #%d: %v", op, uid, err) + writeError(w, http.StatusInternalServerError, "internal error") +} + +// notifyAccount builds the caller's state; it creates the default +// preferences row on first use. A demoted admin's stored admin events are +// not shown (and are not evaluated). +func (s *Server) notifyAccount(u *users.User) (notifyAccountJSON, error) { + n, st, ns := s.auth.notify, s.auth.st, s.auth.set.notify + p, err := st.NotifyPrefsFor(u.ID) + if err != nil { + return notifyAccountJSON{}, err + } + ws, err := st.WatchesFor(u.ID) + if err != nil { + return notifyAccountJSON{}, err + } + pks := make([]string, len(ws)) + for i, w := range ws { + pks[i] = w.Pubkey + } + nodes, err := n.src.nodes(pks, false) + if err != nil { + return notifyAccountJSON{}, err + } + _, perUser, err := st.NotifyMailCounts(n.now().Add(-24 * time.Hour)) + if err != nil { + return notifyAccountJSON{}, err + } + out := notifyAccountJSON{Enabled: p.Enabled, Events: []string{}, AvailableEvents: availableNotifyEvents(u), + Watches: make([]notifyWatchJSON, 0, len(ws)), + Limits: notifyLimitsJSON{MaxWatches: ns.maxWatchesPerUser, PerUserPerDay: ns.perUserPerDay, MailsLast24h: perUser[u.ID]}} + for _, e := range p.Events { + if u.Role == users.RoleAdmin || !users.IsAdminNotifyEvent(e) { + out.Events = append(out.Events, e) + } + } + for _, w := range ws { + nd, known := nodes[w.Pubkey] + out.Watches = append(out.Watches, notifyWatchJSON{Pubkey: w.Pubkey, Name: nd.Name, Known: known, CreatedAt: rfc3339(w.CreatedAt)}) + } + return out, nil +} + +func (s *Server) writeNotifyAccount(w http.ResponseWriter, u *users.User) { + out, err := s.notifyAccount(u) + if err != nil { + notifyInternal(w, "account state", u.ID, err) + return + } + writeJSON(w, out) +} + +func (s *Server) handleNotifyGet(w http.ResponseWriter, _ *http.Request, u *users.User, _ *users.Session) { + s.writeNotifyAccount(w, u) +} + +func (s *Server) handleNotifyPut(w http.ResponseWriter, r *http.Request, u *users.User, _ *users.Session) { + var req notifyPrefsRequest + if !decodeJSON(w, r, &req) { + return + } + for _, e := range req.Events { + if !users.ValidNotifyEvent(e) { + writeError(w, http.StatusBadRequest, "unknown event type") + return + } + if users.IsAdminNotifyEvent(e) && u.Role != users.RoleAdmin { + writeError(w, http.StatusForbidden, "only admins can choose this event") + return + } + } + p, err := s.auth.st.SetNotifyPrefs(u.ID, req.Enabled, req.Events) + if err != nil { + notifyInternal(w, "save preferences", u.ID, err) + return + } + s.auth.audit(idPtr(u.ID), "notify.prefs", idPtr(u.ID), + map[string]string{"enabled": strconv.FormatBool(p.Enabled), "events": strings.Join(p.Events, ",")}) + s.writeNotifyAccount(w, u) +} + +func (s *Server) handleWatchPut(w http.ResponseWriter, r *http.Request, u *users.User, _ *users.Session) { + pk, ok := notifyPubkey(mux.Vars(r)["pubkey"]) + if !ok { + writeError(w, http.StatusBadRequest, "invalid node key: expected 64 hex characters") + return + } + nodes, err := s.auth.notify.src.nodes([]string{pk}, false) + if err != nil { + notifyInternal(w, "look up node", u.ID, err) + return + } + if _, found := nodes[pk]; !found { + writeError(w, http.StatusNotFound, "node not found") + return + } + if _, err := s.auth.st.NotifyPrefsFor(u.ID); err != nil { + notifyInternal(w, "preferences", u.ID, err) + return + } + limit := s.auth.set.notify.maxWatchesPerUser + if err := s.auth.st.AddWatch(u.ID, pk, limit); err != nil { + if errors.Is(err, users.ErrWatchLimit) { + writeError(w, http.StatusConflict, fmt.Sprintf("you watch the maximum of %d nodes; remove one first", limit)) + return + } + notifyInternal(w, "add watch", u.ID, err) + return + } + s.writeNotifyAccount(w, u) +} + +func (s *Server) handleWatchDelete(w http.ResponseWriter, r *http.Request, u *users.User, _ *users.Session) { + pk, ok := notifyPubkey(mux.Vars(r)["pubkey"]) + if !ok { + writeError(w, http.StatusBadRequest, "invalid node key: expected 64 hex characters") + return + } + if err := s.auth.st.RemoveWatch(u.ID, pk); err != nil { + notifyInternal(w, "remove watch", u.ID, err) + return + } + s.writeNotifyAccount(w, u) +} + +// myNodesPubkeys returns the pubkeys of the synced "meshcore-my-nodes" list +// (settings sync; [{pubkey, name, addedAt}]) in list order; nil when the +// document or the key is absent or does not decode. +func myNodesPubkeys(doc string) []string { + if doc == "" { + return nil + } + var d settingsDoc + if json.Unmarshal([]byte(doc), &d) != nil { + return nil + } + raw, ok := d.Keys["meshcore-my-nodes"] + if !ok { + return nil + } + var items []struct { + Pubkey string `json:"pubkey"` + } + if json.Unmarshal([]byte(raw), &items) != nil { + return nil + } + out := make([]string, 0, len(items)) + for _, it := range items { + out = append(out, it.Pubkey) + } + return out +} + +func (s *Server) handleWatchMyNodes(w http.ResponseWriter, _ *http.Request, u *users.User, _ *users.Session) { + st := s.auth.st + doc, _, err := st.GetSettings(u.ID) + if err != nil { + notifyInternal(w, "read synced settings", u.ID, err) + return + } + var res watchMyNodesJSON + seen := map[string]bool{} + var valid []string + for _, raw := range myNodesPubkeys(doc) { + pk, ok := notifyPubkey(raw) + if !ok { + res.Skipped++ + continue + } + if !seen[pk] { + seen[pk] = true + valid = append(valid, pk) + } + } + nodes, err := s.auth.notify.src.nodes(valid, false) + if err != nil { + notifyInternal(w, "look up nodes", u.ID, err) + return + } + known := make([]string, 0, len(valid)) + for _, pk := range valid { + if _, ok := nodes[pk]; ok { + known = append(known, pk) + } else { + res.Skipped++ + } + } + if _, err := st.NotifyPrefsFor(u.ID); err != nil { + notifyInternal(w, "preferences", u.ID, err) + return + } + added, already, over, err := st.AddWatches(u.ID, known, s.auth.set.notify.maxWatchesPerUser) + if err != nil { + notifyInternal(w, "add watches", u.ID, err) + return + } + res.Added, res.Already, res.Skipped = added, already, res.Skipped+over + if res.Account, err = s.notifyAccount(u); err != nil { + notifyInternal(w, "account state", u.ID, err) + return + } + writeJSON(w, res) +} + +// handleUnsubscribeGet sends a mail client or a scanner that follows the +// List-Unsubscribe URL to the confirm view. A GET never changes anything. +func (s *Server) handleUnsubscribeGet(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, s.auth.link("unsubscribe", r.URL.Query().Get("token")), http.StatusSeeOther) +} + +// handleUnsubscribePost turns notifications off for the token's owner: the +// confirm button and the provider's one-click POST. The token only does this. +func (s *Server) handleUnsubscribePost(w http.ResponseWriter, r *http.Request) { + uid, was, err := s.auth.st.DisableNotifyByToken(r.URL.Query().Get("token")) + if err != nil { + if !errors.Is(err, users.ErrTokenInvalid) { + log.Printf("[notify] unsubscribe: %v", err) + } + writeTokenError(w, err) + return + } + if was { + s.auth.audit(idPtr(uid), "notify.unsubscribe", idPtr(uid), map[string]string{"via": "link"}) + } + writeJSON(w, okResponse{OK: true, Message: "Notifications are off. Turn them back on from your account page."}) +} diff --git a/cmd/server/notify_handlers_test.go b/cmd/server/notify_handlers_test.go new file mode 100644 index 00000000..732dd8ad --- /dev/null +++ b/cmd/server/notify_handlers_test.go @@ -0,0 +1,273 @@ +package main + +import ( + "encoding/json" + "net/http/httptest" + "reflect" + "strings" + "testing" + "time" + + "github.com/meshcore-analyzer/users" +) + +func notifyPath(pk string) string { return "/api/account/notifications/watches/" + pk } + +func TestNotifyRoutesAbsentWhenOff(t *testing.T) { + f := newNotifyFixture(t, notifySettings{}) + for _, rt := range []struct{ method, path string }{ + {"GET", "/api/account/notifications"}, {"PUT", "/api/account/notifications"}, + {"PUT", notifyPath(evPkA)}, {"DELETE", notifyPath(evPkA)}, + {"POST", "/api/account/notifications/watch-my-nodes"}, + {"GET", "/api/notifications/unsubscribe?token=x"}, {"POST", "/api/notifications/unsubscribe?token=x"}, + } { + if w := f.do(rt.method, rt.path, nil); w.Code != 404 { + t.Errorf("%s %s = %d with notifications off; want 404", rt.method, rt.path, w.Code) + } + } +} + +func TestNotifyGetDefaultsAndAdminEvents(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + boss := f.registerAndActivate(t, "admin@example.org", "Boss", pw) + pat := f.registerAndActivate(t, "pat@example.org", "Pat", pw) + expectStatus(t, f.do("GET", "/api/account/notifications", nil), 401) + w := f.do("GET", "/api/account/notifications", nil, as(pat)) + expectStatus(t, w, 200) + want := notifyAccountJSON{Enabled: true, Events: []string{"node.offline", "node.battery"}, AvailableEvents: []string{"node.offline", "node.battery"}, + Watches: []notifyWatchJSON{}, Limits: notifyLimitsJSON{MaxWatches: 50, PerUserPerDay: 20}} + if got := decode[notifyAccountJSON](t, w); !reflect.DeepEqual(got, want) { + t.Fatalf("user defaults = %+v; want %+v", got, want) + } + if !strings.Contains(w.Body.String(), `"watches":[]`) { + t.Fatalf("empty watch list is not an array: %s", w.Body.String()) + } + adm := decode[notifyAccountJSON](t, f.do("GET", "/api/account/notifications", nil, as(boss))) + if !reflect.DeepEqual(adm.AvailableEvents, []string{"node.offline", "node.battery", "foreign.new", "observer.offline"}) || + !reflect.DeepEqual(adm.Events, []string{"node.offline", "node.battery"}) { + t.Fatalf("admin = %+v", adm) + } +} + +func TestNotifyPutPrefs(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + boss := f.registerAndActivate(t, "admin@example.org", "Boss", pw) + pat := f.registerAndActivate(t, "pat@example.org", "Pat", pw) + put := func(c *client, body any) *httptest.ResponseRecorder { + return f.do("PUT", "/api/account/notifications", body, as(c)) + } + expectStatus(t, put(&client{cookie: pat.cookie}, notifyPrefsRequest{Enabled: true}), 403) + w := put(pat, notifyPrefsRequest{Enabled: false, Events: []string{"node.battery"}}) + expectStatus(t, w, 200) + if got := decode[notifyAccountJSON](t, w); got.Enabled || !reflect.DeepEqual(got.Events, []string{"node.battery"}) { + t.Fatalf("after PUT = %+v", got) + } + entries, err := f.st.AuditList(users.AuditFilter{Actions: []string{"notify.prefs"}}) + if err != nil || len(entries) != 1 || *entries[0].ActorUserID != pat.me.ID || entries[0].Detail["enabled"] != "false" || entries[0].Detail["events"] != "node.battery" { + t.Fatalf("audit = %+v, %v", entries, err) + } + if w := put(pat, notifyPrefsRequest{Enabled: true, Events: []string{"node.reboot"}}); w.Code != 400 || !strings.Contains(w.Body.String(), "unknown event type") { + t.Errorf("unknown event: %d %s", w.Code, w.Body.String()) + } + if w := put(pat, notifyPrefsRequest{Enabled: true, Events: []string{"foreign.new"}}); w.Code != 403 || !strings.Contains(w.Body.String(), "only admins") { + t.Errorf("admin event by a user: %d %s", w.Code, w.Body.String()) + } + if w := put(pat, map[string]any{"enabled": true, "x": 1}); w.Code != 400 { + t.Errorf("unknown field: %d", w.Code) + } + w = put(boss, notifyPrefsRequest{Enabled: true, Events: []string{"observer.offline", "foreign.new"}}) + if got := decode[notifyAccountJSON](t, w); w.Code != 200 || !reflect.DeepEqual(got.Events, []string{"foreign.new", "observer.offline"}) { + t.Fatalf("admin PUT: %d %+v", w.Code, got) + } +} + +func TestWatchPutAndDelete(t *testing.T) { + ns := defaultNotifySettings() + ns.maxWatchesPerUser = 1 + f := newNotifyFixture(t, ns) + f.setNode(evPkA, "Alpha", "repeater", time.Hour, nil) + f.setNode(evPkB, "Bravo", "companion", time.Hour, nil) + pat := f.registerAndActivate(t, "pat@example.org", "Pat", pw) + expectStatus(t, f.do("PUT", notifyPath(evPkA), nil), 401) + expectStatus(t, f.do("PUT", notifyPath(evPkA), nil, as(&client{cookie: pat.cookie})), 403) + for _, bad := range []string{"xyz", evPkA[:63], evPkA + "0", strings.Repeat("g", 64)} { + if w := f.do("PUT", notifyPath(bad), nil, as(pat)); w.Code != 400 { + t.Errorf("PUT %q = %d; want 400", bad, w.Code) + } + } + if w := f.do("PUT", notifyPath(evPkG), nil, as(pat)); w.Code != 404 || !strings.Contains(w.Body.String(), "node not found") { + t.Fatalf("unknown node: %d %s", w.Code, w.Body.String()) + } + w := f.do("PUT", notifyPath(strings.ToUpper(evPkA)), nil, as(pat)) + expectStatus(t, w, 200) + got := decode[notifyAccountJSON](t, w) + if len(got.Watches) != 1 || got.Watches[0].Pubkey != evPkA || got.Watches[0].Name != "Alpha" || !got.Watches[0].Known { + t.Fatalf("after watching = %+v", got.Watches) + } + expectStatus(t, f.do("PUT", notifyPath(evPkA), nil, as(pat)), 200) + if w := f.do("PUT", notifyPath(evPkB), nil, as(pat)); w.Code != 409 || !strings.Contains(w.Body.String(), "you watch the maximum of 1 nodes") { + t.Fatalf("over the limit: %d %s", w.Code, w.Body.String()) + } + delete(f.src.nodeMap, evPkA) // the node left the analyzer database + if got := decode[notifyAccountJSON](t, f.do("GET", "/api/account/notifications", nil, as(pat))); got.Watches[0].Known || got.Watches[0].Name != "" { + t.Fatalf("vanished node = %+v", got.Watches[0]) + } + w = f.do("DELETE", notifyPath(evPkA), nil, as(pat)) + expectStatus(t, w, 200) + if got := decode[notifyAccountJSON](t, w); len(got.Watches) != 0 { + t.Fatalf("after DELETE = %+v", got.Watches) + } + expectStatus(t, f.do("DELETE", notifyPath(evPkA), nil, as(pat)), 200) + expectStatus(t, f.do("DELETE", notifyPath("xyz"), nil, as(pat)), 400) + if entries, _ := f.st.AuditList(users.AuditFilter{Actions: []string{"notify.*"}}); len(entries) != 0 { + t.Fatalf("watch changes were audited: %+v", entries) + } +} + +func TestMyNodesPubkeys(t *testing.T) { + for _, doc := range []string{"", "not json", `{"v":1,"keys":{}}`, `{"v":1,"keys":{"meshcore-my-nodes":"not json"}}`} { + if got := myNodesPubkeys(doc); got != nil { + t.Errorf("%q = %v; want nil", doc, got) + } + } + got := myNodesPubkeys(`{"v":1,"keys":{"meshcore-my-nodes":"[{\"pubkey\":\"ab\",\"name\":\"x\",\"addedAt\":1},{\"name\":\"no key\"}]"}}`) + if !reflect.DeepEqual(got, []string{"ab", ""}) { + t.Fatalf("myNodesPubkeys = %v", got) + } +} + +func TestWatchMyNodes(t *testing.T) { + ns := defaultNotifySettings() + ns.maxWatchesPerUser = 1 + f := newNotifyFixture(t, ns) + f.setNode(evPkA, "Alpha", "repeater", time.Hour, nil) + f.setNode(evPkB, "Bravo", "companion", time.Hour, nil) + pat := f.registerAndActivate(t, "pat@example.org", "Pat", pw) + post := func() watchMyNodesJSON { + t.Helper() + w := f.do("POST", "/api/account/notifications/watch-my-nodes", nil, as(pat)) + expectStatus(t, w, 200) + return decode[watchMyNodesJSON](t, w) + } + if r := post(); r.Added != 0 || r.Already != 0 || r.Skipped != 0 { + t.Fatalf("without synced settings = %+v", r) + } + list, _ := json.Marshal([]map[string]string{{"pubkey": strings.ToUpper(evPkA)}, {"pubkey": evPkB}, {"pubkey": evPkG}, {"pubkey": "bad"}, {"pubkey": evPkA}}) + doc, _ := json.Marshal(settingsDoc{V: 1, Keys: map[string]string{"meshcore-my-nodes": string(list)}}) + if _, err := f.st.PutSettings(pat.me.ID, users.SettingsVersion{}, string(doc)); err != nil { + t.Fatal(err) + } + // A watched, B over the limit, G not in the analyzer, "bad" malformed, the second A a duplicate. + if r := post(); r.Added != 1 || r.Already != 0 || r.Skipped != 3 || len(r.Account.Watches) != 1 || r.Account.Watches[0].Pubkey != evPkA { + t.Fatalf("first copy = %+v", r) + } + if r := post(); r.Added != 0 || r.Already != 1 || r.Skipped != 3 { + t.Fatalf("second copy = %+v", r) + } + expectStatus(t, f.do("POST", "/api/account/notifications/watch-my-nodes", nil, as(&client{cookie: pat.cookie})), 403) +} + +func TestUnsubscribe(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + pat := f.registerAndActivate(t, "pat@example.org", "Pat", pw) + p, err := f.st.NotifyPrefsFor(pat.me.ID) + if err != nil { + t.Fatal(err) + } + q := "/api/notifications/unsubscribe?token=" + p.UnsubToken + w := f.do("GET", q, nil) + if w.Code != 303 || w.Header().Get("Location") != testBase+"/#/account/unsubscribe?token="+p.UnsubToken { + t.Fatalf("GET = %d, Location %q", w.Code, w.Header().Get("Location")) + } + if again, _ := f.st.NotifyPrefsFor(pat.me.ID); !again.Enabled { + t.Fatal("a GET (a link scanner) turned notifications off") + } + // The provider's one-click POST: form body, a foreign Origin, no session. + w = f.do("POST", q, nil, header("Origin", "https://mail.example.net"), header("Content-Type", "application/x-www-form-urlencoded")) + expectStatus(t, w, 200) + if r := decode[okResponse](t, w); !r.OK || !strings.Contains(r.Message, "Notifications are off") { + t.Fatalf("POST = %+v", r) + } + if again, _ := f.st.NotifyPrefsFor(pat.me.ID); again.Enabled { + t.Fatal("still enabled after the POST") + } + expectStatus(t, f.do("POST", q, nil), 200) + entries, err := f.st.AuditList(users.AuditFilter{Actions: []string{"notify.unsubscribe"}}) + if err != nil || len(entries) != 1 || *entries[0].TargetUserID != pat.me.ID || entries[0].Detail["via"] != "link" { + t.Fatalf("audit = %+v, %v; want one row for the change", entries, err) + } + for _, bad := range []string{"/api/notifications/unsubscribe?token=nope", "/api/notifications/unsubscribe"} { + if w := f.do("POST", bad, nil); w.Code != 410 { + t.Errorf("POST %s = %d; want 410", bad, w.Code) + } + } +} + +func TestClientConfigAdvertisesNotifications(t *testing.T) { + srv, router := setupTestServer(t) + a, _ := newTestAuthService(t) + a.notify = newNotifier(a, &fakeNotifySource{}, time.Now) + srv.auth = a + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", "/api/config/client", nil)) + if !strings.Contains(w.Body.String(), `"userManagement":{"enabled":true,"notifications":true}`) { + t.Fatalf("client config = %s", w.Body.String()) + } +} + +func TestAdminStatsNotifyBlock(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + boss := f.registerAndActivate(t, "admin@example.org", "Boss", pw) + f.watcher(t, "pat@example.org", "Pat", evPkA, evPkB) + if _, err := f.st.LogMail(nil, "gone@example.org", users.NotifyMailPurpose, ""); err != nil { + t.Fatal(err) + } + w := f.do("GET", "/api/admin/stats", nil, as(boss)) + expectStatus(t, w, 200) + st := decode[adminStatsJSON](t, w) + if st.Notify == nil || *st.Notify != (adminNotifyJSON{MailsLast24h: 1, MaxMailsPerDay: 300, Watches: 2, WatchingUsers: 1}) { + t.Fatalf("notify block = %+v", st.Notify) + } + off, offBoss, _ := adminFixture(t) + if w := off.do("GET", "/api/admin/stats", nil, as(offBoss)); strings.Contains(w.Body.String(), `"notify"`) { + t.Fatalf("notify block with notifications off: %s", w.Body.String()) + } +} + +// users.db never holds case variants: every route lowercases the key first. +func TestWatchRoutesStoreLowercaseKeys(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.setNode(evPkA, "Alpha", "repeater", time.Hour, nil) + f.setNode(evPkB, "Bravo", "companion", time.Hour, nil) + pat := f.registerAndActivate(t, "pat@example.org", "Pat", pw) + stored := func() []string { + t.Helper() + ws, err := f.st.WatchesFor(pat.me.ID) + if err != nil { + t.Fatal(err) + } + out := []string{} + for _, w := range ws { + out = append(out, w.Pubkey) + } + return out + } + expectStatus(t, f.do("PUT", notifyPath(strings.ToUpper(evPkA)), nil, as(pat)), 200) + expectStatus(t, f.do("PUT", notifyPath(evPkA), nil, as(pat)), 200) + if got := stored(); !reflect.DeepEqual(got, []string{evPkA}) { + t.Fatalf("after PUT upper then lower = %v", got) + } + expectStatus(t, f.do("DELETE", notifyPath(strings.ToUpper(evPkA)), nil, as(pat)), 200) + if got := stored(); len(got) != 0 { + t.Fatalf("DELETE with an uppercase key left %v", got) + } + list, _ := json.Marshal([]map[string]string{{"pubkey": strings.ToUpper(evPkB)}}) + doc, _ := json.Marshal(settingsDoc{V: 1, Keys: map[string]string{"meshcore-my-nodes": string(list)}}) + if _, err := f.st.PutSettings(pat.me.ID, users.SettingsVersion{}, string(doc)); err != nil { + t.Fatal(err) + } + expectStatus(t, f.do("POST", "/api/account/notifications/watch-my-nodes", nil, as(pat)), 200) + if got := stored(); !reflect.DeepEqual(got, []string{evPkB}) { + t.Fatalf("watch-my-nodes stored %v", got) + } +} diff --git a/cmd/server/notify_service.go b/cmd/server/notify_service.go new file mode 100644 index 00000000..4e3d2282 --- /dev/null +++ b/cmd/server/notify_service.go @@ -0,0 +1,357 @@ +package main + +import ( + "context" + "fmt" + "log" + "net/url" + "sort" + "strings" + "time" + + "github.com/meshcore-analyzer/mailer" + "github.com/meshcore-analyzer/users" +) + +// notifier evaluates node notifications on a timer and mails the changes +// (docs/specs/2026-10-07-node-notifications-design.md). authService.notify +// is nil unless userManagement.notifications.enabled. +type notifier struct { + a *authService + src notifySource + now func() time.Time + // ingestStale is the last tick's verdict, so the pause and the resume + // are each logged once; ingestResumedAt is when the last stale period + // ended (the offline grace, see notifyInput). In memory only: a restart + // inside the grace loses it. Only the loop goroutine touches them. + ingestStale bool + ingestResumedAt time.Time +} + +func newNotifier(a *authService, src notifySource, now func() time.Time) *notifier { + return ¬ifier{a: a, src: src, now: now} +} + +// notifySendTimeout bounds one mail send inside a tick. +const notifySendTimeout = 30 * time.Second + +// notifyIngestStaleAfter: when the newest packet in the store is older, +// offline checks pause (an MQTT or ingestor outage would otherwise mail +// every watcher "offline" and later "back online"). +const notifyIngestStaleAfter = 30 * time.Minute + +// checkIngest reports whether ingest is stale at now and logs when that +// verdict changes. +func (n *notifier) checkIngest(now time.Time) bool { + newest := n.src.newestPacket() + stale := newest.IsZero() || now.Sub(newest) > notifyIngestStaleAfter + switch { + case stale && !n.ingestStale: + since := "(no packets)" + if !newest.IsZero() { + since = newest.UTC().Format(time.RFC3339) + } + log.Printf("[notify] ingest stale since %s; offline checks paused", since) + case !stale && n.ingestStale: + n.ingestResumedAt = now + log.Printf("[notify] ingest fresh again; offline checks resumed") + } + n.ingestStale = stale + return stale +} + +// loop evaluates every interval until stop closes. The first evaluation +// runs one interval after startup. A tick in flight sees its context +// cancelled when stop closes, so a slow mail provider does not hold up +// the shutdown. +func (n *notifier) loop(stop <-chan struct{}, every time.Duration) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + select { + case <-stop: + cancel() + case <-ctx.Done(): + } + }() + t := time.NewTicker(every) + defer t.Stop() + for { + select { + case <-stop: + return + case <-t.C: + n.safeTick(ctx) + } + } +} + +// safeTick runs one evaluation; a panic is logged and the next tick runs. +func (n *notifier) safeTick(ctx context.Context) { + defer func() { + if r := recover(); r != nil { + log.Printf("[notify] evaluation panicked, the next one runs as planned: %v", r) + } + }() + n.tick(ctx) +} + +// tick reads all preferences, watches and states (one query each), the +// accounts and the analyzer snapshot they need, evaluates, deletes the +// dropped states and writes the new ones, and then mails each user's +// changes. Users without a preferences row and users with notifications +// off are evaluated too (the evaluator gives the first the defaults and +// keeps the second's states current without changes). A failed read or +// state write ends the tick without mail; the next tick starts over. A +// tick that evaluates logs one line with its counts, its duration and how +// long LastHeardMap held the packet store's read lock. +func (n *notifier) tick(ctx context.Context) { + if !n.src.ready() { + return + } + started := time.Now() + now := n.now() + stale := n.checkIngest(now) + st := n.a.st + prefs, err := st.AllNotifyPrefs() + if err != nil { + log.Printf("[notify] read preferences: %v", err) + return + } + stored, err := st.AllWatches() + if err != nil { + log.Printf("[notify] read watches: %v", err) + return + } + if len(prefs) == 0 && len(stored) == 0 { + return + } + states, err := st.AllNotifyStates() + if err != nil { + log.Printf("[notify] read states: %v", err) + return + } + // The analyzer snapshot is keyed by lowercase pubkey; normalise the + // watches once so every lookup and state subject uses that form. + watches := make([]users.NotifyWatch, 0, len(stored)) + watchSeen := make(map[users.NotifyKey]bool, len(stored)) + seen := map[string]bool{} + var pubkeys []string + for _, w := range stored { + w.Pubkey = strings.ToLower(w.Pubkey) + k := users.NotifyKey{UserID: w.UserID, Subject: w.Pubkey} + if watchSeen[k] { + continue + } + watchSeen[k] = true + watches = append(watches, w) + if !seen[w.Pubkey] { + seen[w.Pubkey] = true + pubkeys = append(pubkeys, w.Pubkey) + } + } + byUser := make(map[int64]users.NotifyPrefs, len(prefs)) + var ids []int64 + for _, p := range prefs { + byUser[p.UserID] = p + ids = append(ids, p.UserID) + } + noPrefs := map[int64]bool{} + for _, w := range watches { + if _, ok := byUser[w.UserID]; !ok && !noPrefs[w.UserID] { + noPrefs[w.UserID] = true + ids = append(ids, w.UserID) + } + } + accounts, err := st.UsersByID(ids) + if err != nil { + log.Printf("[notify] read accounts: %v", err) + return + } + wantForeign, wantObservers := false, false + for _, p := range prefs { + if accounts[p.UserID].Role != users.RoleAdmin { + continue + } + wantForeign = wantForeign || p.Has(users.NotifyForeignNew) + wantObservers = wantObservers || p.Has(users.NotifyObserverOffline) + } + nodes, err := n.src.nodes(pubkeys, wantForeign) + if err != nil { + log.Printf("[notify] read nodes: %v", err) + return + } + var infra []string + for _, pk := range pubkeys { + if nd, ok := nodes[pk]; ok && notifyInfra(nd.Role) { + infra = append(infra, pk) + } + } + var observers []notifyObserver + if wantObservers { + if observers, err = n.src.observers(); err != nil { + log.Printf("[notify] read observers: %v", err) + return + } + } + heard, lock := n.src.lastHeard(pubkeys) + res := evaluateNotifications(notifyInput{Now: now, Health: n.src.health(), LowMv: n.src.lowBatteryMv(), + Accounts: accounts, Prefs: prefs, Watches: watches, States: states, Nodes: nodes, + Heard: heard, Relayed: n.src.lastRelayed(infra), Observers: observers, IngestStale: stale, IngestResumedAt: n.ingestResumedAt}) + if err := st.DeleteNotifyStates(res.Drop); err != nil { + log.Printf("[notify] delete dropped states, nothing mailed this time: %v", err) + return + } + if len(res.States) > 0 { + if err := st.WriteNotifyStates(res.States); err != nil { + log.Printf("[notify] write states, nothing mailed this time: %v", err) + return + } + } + mails := n.deliver(ctx, now, res.Changes, accounts, byUser) + changes := 0 + for _, list := range res.Changes { + changes += len(list) + } + log.Printf("[notify] tick: users=%d changes=%d mails=%d took=%.2fms lock=%.2fms", + len(accounts), changes, mails, msFloat(time.Since(started)), msFloat(lock)) +} + +func msFloat(d time.Duration) float64 { return float64(d.Microseconds()) / 1000 } + +// notifySkipReason says why a user's changes are not mailed; "" means send. +// Skipped changes are already stored and are never mailed later. +func notifySkipReason(u users.User, p users.NotifyPrefs, userMails, totalMails int, ns notifySettings) string { + switch { + case u.Status != users.StatusActive: + return "account not active" + case u.EmailBouncing: + return "address bounces" + case !p.Enabled: + return "notifications off" + case userMails >= ns.perUserPerDay: + return "daily limit for this user reached" + case totalMails >= ns.maxMailsPerDay: + return "daily limit for this instance reached" + } + return "" +} + +// deliver sends one mail per user with changes, within the limits counted +// from mail_log over the last 24 hours (this tick's mails included). A user +// with changes but no preferences row was evaluated with the defaults; the +// row (and so the unsubscribe token) is created before the first mail. +// Once ctx is cancelled (shutdown) the remaining users are not mailed; +// their changes are already stored, like any other skipped change. It +// returns the number of mails sent. +func (n *notifier) deliver(ctx context.Context, now time.Time, changes map[int64][]notifyChange, + accounts map[int64]users.User, prefs map[int64]users.NotifyPrefs) int { + if len(changes) == 0 { + return 0 + } + total, perUser, err := n.a.st.NotifyMailCounts(now.Add(-24 * time.Hour)) + if err != nil { + log.Printf("[notify] count mails, nothing mailed this time: %v", err) + return 0 + } + sent := 0 + ids := make([]int64, 0, len(changes)) + for uid := range changes { + ids = append(ids, uid) + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + for i, uid := range ids { + if ctx.Err() != nil { + log.Printf("[notify] shutting down: %d user(s) with changes not mailed", len(ids)-i) + return sent + } + u := accounts[uid] + p, ok := prefs[uid] + if !ok { + p = users.DefaultNotifyPrefs(uid) + } + if why := notifySkipReason(u, p, perUser[uid], total, n.a.set.notify); why != "" { + log.Printf("[notify] user #%d: %d change(s) not mailed: %s", uid, len(changes[uid]), why) + continue + } + if !ok { + if p, err = n.a.st.NotifyPrefsFor(uid); err != nil { + log.Printf("[notify] user #%d: create preferences, %d change(s) not mailed: %v", uid, len(changes[uid]), err) + continue + } + } + sendCtx, cancel := context.WithTimeout(ctx, notifySendTimeout) + err := n.a.sendMail(sendCtx, &u, users.NotifyMailPurpose, n.a.notifyMail(u, p, changes[uid])) + cancel() + if err != nil { + continue // logged by sendMail; the changes stay recorded + } + total++ + perUser[uid]++ + sent++ + } + return sent +} + +// notifyMail is the one mail for all of a user's changes in one evaluation. +func (a *authService) notifyMail(u users.User, p users.NotifyPrefs, changes []notifyChange) mailer.Message { + base := a.set.baseURL.String() + lines := make([]mailLine, 0, len(changes)) + for _, c := range changes { + lines = append(lines, mailLine{text: notifyChangeText(c), url: notifySubjectURL(base, c)}) + } + subject := fmt.Sprintf("%d changes on your watched nodes", len(changes)) + if len(changes) == 1 { + subject = "1 change on your watched nodes" + } + msg := a.render(u.Email, u.DisplayName, "notify", mailContent{ + subject: subject, greeting: "Hello " + u.DisplayName + ",", + paragraphs: []string{"Changes since the last check:"}, + lines: lines, + actionLabel: "Manage notifications", actionURL: base + "/#/account?section=notifications", + footer: &mailLine{text: "You get this mail because you asked " + a.set.baseURL.Host + + " to watch these nodes. Turn all notification mails off with one click:", url: a.link("unsubscribe", p.UnsubToken)}, + }) + msg.Headers = map[string]string{ + "List-Unsubscribe": "<" + base + "/api/notifications/unsubscribe?token=" + url.QueryEscape(p.UnsubToken) + ">", + "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", + } + return msg +} + +// notifyChangeText is one line of the mail: name, what happened, when (UTC). +func notifyChangeText(c notifyChange) string { + bad := c.To == users.NotifyBad + var what string + switch c.Event { + case users.NotifyNodeOffline: + what = "back online" + if bad { + what = "offline" + } + case users.NotifyNodeBattery: + what = "battery recovered" + if bad { + what = "battery low" + } + if c.BatteryMv != nil { + what += fmt.Sprintf(" (%d mV)", *c.BatteryMv) + } + case users.NotifyForeignNew: + what = "new foreign node" + case users.NotifyObserverOffline: + what = "observer back online" + if bad { + what = "observer offline" + } + } + return c.Name + ": " + what + ", " + c.At.UTC().Format("2006-01-02 15:04 UTC") +} + +// notifySubjectURL links a change to its node or observer page. +func notifySubjectURL(base string, c notifyChange) string { + if c.Event == users.NotifyObserverOffline { + return base + "/#/observers/" + url.PathEscape(c.Subject) + } + return base + "/#/nodes/" + url.PathEscape(c.Subject) +} diff --git a/cmd/server/notify_service_test.go b/cmd/server/notify_service_test.go new file mode 100644 index 00000000..c9b8b32f --- /dev/null +++ b/cmd/server/notify_service_test.go @@ -0,0 +1,727 @@ +package main + +import ( + "context" + "errors" + "path/filepath" + "regexp" + "strings" + "testing" + "time" + + "github.com/gorilla/mux" + "github.com/meshcore-analyzer/mailer" + "github.com/meshcore-analyzer/users" +) + +type fakeNotifySource struct { + isReady bool + hs HealthThresholds + low int + nodeMap map[string]notifyNode + heard map[string]time.Time + relayed map[string]time.Time + obs []notifyObserver + newestAt func() time.Time // newest packet in the store; nil means none + lock time.Duration // what lastHeard reports as its lock time + panicNext bool + heardAsked []string + foreignAsked int + observersAsked int +} + +func (f *fakeNotifySource) ready() bool { return f.isReady } +func (f *fakeNotifySource) health() HealthThresholds { return f.hs } +func (f *fakeNotifySource) lowBatteryMv() int { return f.low } +func (f *fakeNotifySource) nodes(pubkeys []string, withForeign bool) (map[string]notifyNode, error) { + if f.panicNext { + f.panicNext = false + panic("fake source panic") + } + out := map[string]notifyNode{} + for _, pk := range pubkeys { + if n, ok := f.nodeMap[pk]; ok { + out[pk] = n + } + } + if withForeign { + f.foreignAsked++ + for pk, n := range f.nodeMap { + if n.Foreign { + out[pk] = n + } + } + } + return out, nil +} +func (f *fakeNotifySource) lastHeard(pks []string) (map[string]time.Time, time.Duration) { + f.heardAsked = append([]string(nil), pks...) + return f.heard, f.lock +} +func (f *fakeNotifySource) lastRelayed([]string) map[string]time.Time { return f.relayed } +func (f *fakeNotifySource) newestPacket() time.Time { + if f.newestAt == nil { + return time.Time{} + } + return f.newestAt() +} +func (f *fakeNotifySource) observers() ([]notifyObserver, error) { + f.observersAsked++ + return f.obs, nil +} + +type notifyClock struct{ t time.Time } + +func (c *notifyClock) Now() time.Time { return c.t } +func (c *notifyClock) Advance(d time.Duration) { c.t = c.t.Add(d) } + +func defaultNotifySettings() notifySettings { + return notifySettings{enabled: true, interval: 5 * time.Minute, perUserPerDay: 20, maxMailsPerDay: 300, maxWatchesPerUser: 50} +} + +type notifyFixture struct { + *authFixture + n *notifier + src *fakeNotifySource + clk *notifyClock +} + +// newNotifyFixture is newAuthFixture with notifications set to ns, a fake +// analyzer source and one fixed clock for the notifier and users.db. +// admin@example.org is a config admin. +func newNotifyFixture(t *testing.T, ns notifySettings) *notifyFixture { + t.Helper() + a, fake := newTestAuthService(t, "admin@example.org") + a.set.notify = ns + clk := ¬ifyClock{t: time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)} + a.st.SetClock(clk.Now) + src := &fakeNotifySource{isReady: true, hs: (&Config{}).GetHealthThresholds(), low: 3300, + nodeMap: map[string]notifyNode{}, heard: map[string]time.Time{}, relayed: map[string]time.Time{}, newestAt: clk.Now} + if ns.enabled { + a.notify = newNotifier(a, src, clk.Now) + } + srv := &Server{cfg: &Config{APIKey: testAPIKey}, perfStats: NewPerfStats(), auth: a} + r := mux.NewRouter() + srv.registerAuthRoutes(r) + return ¬ifyFixture{authFixture: &authFixture{srv: srv, router: r, fake: fake, st: a.st}, n: a.notify, src: src, clk: clk} +} + +// watcher registers and activates a user with default prefs watching pubkeys. +func (f *notifyFixture) watcher(t *testing.T, email, name string, pubkeys ...string) *client { + t.Helper() + c := f.registerAndActivate(t, email, name, pw) + if _, err := f.st.NotifyPrefsFor(c.me.ID); err != nil { + t.Fatal(err) + } + if _, _, _, err := f.st.AddWatches(c.me.ID, pubkeys, 0); err != nil { + t.Fatal(err) + } + return c +} + +// setNode puts a node in the fake analyzer, last seen lastSeen before now. +func (f *notifyFixture) setNode(pk, name, role string, lastSeen time.Duration, battery *int) { + f.src.nodeMap[pk] = notifyNode{Pubkey: pk, Name: name, Role: role, LastSeen: f.clk.t.Add(-lastSeen).Format(time.RFC3339), BatteryMv: battery} +} + +func (f *notifyFixture) tick() { f.n.tick(context.Background()) } + +func (f *notifyFixture) notifyMails() []mailer.Message { + var out []mailer.Message + for _, m := range f.fake.Sent() { + if m.Tag == "notify" { + out = append(out, m) + } + } + return out +} + +func TestNotifierMailsAChangeAfterTheSilentFirstTick(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + pat := f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.tick() + if m := f.notifyMails(); len(m) != 0 { + t.Fatalf("first tick mailed: %+v", m) + } + f.clk.Advance(25 * time.Hour) + f.tick() + m := f.notifyMails() + if len(m) != 1 { + t.Fatalf("mails = %d; want 1", len(m)) + } + p, _ := f.st.NotifyPrefsFor(pat.me.ID) + if m[0].To != "pat@example.org" || m[0].Subject != "[CoreScope] 1 change on your watched nodes" || + !strings.Contains(m[0].Text, "Alpha: offline, 2026-10-08 13:00 UTC") || + !strings.Contains(m[0].Text, testBase+"/#/nodes/"+evPkA) || + !strings.Contains(m[0].Text, testBase+"/#/account?section=notifications") || + !strings.Contains(m[0].Text, testBase+"/#/account/unsubscribe?token="+p.UnsubToken) { + t.Fatalf("mail = %+v", m[0]) + } + if m[0].Headers["List-Unsubscribe"] != "<"+testBase+"/api/notifications/unsubscribe?token="+p.UnsubToken+">" || + m[0].Headers["List-Unsubscribe-Post"] != "List-Unsubscribe=One-Click" { + t.Fatalf("headers = %v", m[0].Headers) + } + if total, per, err := f.st.NotifyMailCounts(f.clk.t.Add(-time.Hour)); err != nil || total != 1 || per[pat.me.ID] != 1 { + t.Fatalf("mail_log notify rows = %d, %v, %v", total, per, err) + } + f.clk.Advance(5 * time.Minute) + f.tick() + if len(f.notifyMails()) != 1 { + t.Fatal("an unchanged state was mailed again") + } + f.src.heard[evPkA] = f.clk.t.Add(-time.Minute) + f.tick() + if m := f.notifyMails(); len(m) != 2 || !strings.Contains(m[1].Text, "Alpha: back online") { + t.Fatalf("back-online mail = %+v", m) + } +} + +func TestNotifierOneMailPerUserPerTick(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA, evPkB) + f.watcher(t, "quinn@example.org", "Quinn", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.setNode(evPkB, "Bravo", "companion", time.Hour, nil) + f.tick() + f.clk.Advance(25 * time.Hour) + f.tick() + m := f.notifyMails() + if len(m) != 2 { + t.Fatalf("mails = %d; want one per user", len(m)) + } + byTo := map[string]mailer.Message{m[0].To: m[0], m[1].To: m[1]} + pat, quinn := byTo["pat@example.org"], byTo["quinn@example.org"] + if pat.Subject != "[CoreScope] 2 changes on your watched nodes" || !strings.Contains(pat.Text, "Alpha: offline") || !strings.Contains(pat.Text, "Bravo: offline") { + t.Fatalf("Pat's mail = %+v", pat) + } + if quinn.Subject != "[CoreScope] 1 change on your watched nodes" || strings.Contains(quinn.Text, "Bravo") { + t.Fatalf("Quinn's mail = %+v", quinn) + } +} + +func TestNotifierSkipsAreNeverMailedLater(t *testing.T) { + cases := []struct { + name string + tune func(*notifySettings) + block func(*notifyFixture, int64) error + lift func(*notifyFixture, int64) error + }{ + {"notifications off", nil, + func(f *notifyFixture, uid int64) error { + _, err := f.st.SetNotifyPrefs(uid, false, users.NodeNotifyEvents) + return err + }, + func(f *notifyFixture, uid int64) error { + _, err := f.st.SetNotifyPrefs(uid, true, users.NodeNotifyEvents) + return err + }}, + {"address bounces", nil, + func(f *notifyFixture, uid int64) error { return f.st.SetEmailBouncing(uid, true) }, + func(f *notifyFixture, uid int64) error { return f.st.SetEmailBouncing(uid, false) }}, + {"account disabled", nil, + func(f *notifyFixture, uid int64) error { return f.st.SetStatus(uid, users.StatusDisabled) }, + func(f *notifyFixture, uid int64) error { return f.st.SetStatus(uid, users.StatusActive) }}, + {"per-user limit", func(ns *notifySettings) { ns.perUserPerDay = 1 }, + func(f *notifyFixture, uid int64) error { + _, err := f.st.LogMail(&uid, "pat@example.org", users.NotifyMailPurpose, "") + return err + }, + func(f *notifyFixture, uid int64) error { f.clk.Advance(25 * time.Hour); return nil }}, + {"instance limit", func(ns *notifySettings) { ns.maxMailsPerDay = 1 }, + func(f *notifyFixture, uid int64) error { + _, err := f.st.LogMail(nil, "gone@example.org", users.NotifyMailPurpose, "") + return err + }, + func(f *notifyFixture, uid int64) error { f.clk.Advance(25 * time.Hour); return nil }}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + ns := defaultNotifySettings() + if c.tune != nil { + c.tune(&ns) + } + f := newNotifyFixture(t, ns) + pat := f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.tick() + // Advance first, so a blocking mail_log row lies inside the + // rolling 24 hours of the next tick. + f.clk.Advance(25 * time.Hour) + if err := c.block(f, pat.me.ID); err != nil { + t.Fatal(err) + } + f.tick() + if m := f.notifyMails(); len(m) != 0 { + t.Fatalf("mailed while blocked: %+v", m) + } + if err := c.lift(f, pat.me.ID); err != nil { + t.Fatal(err) + } + f.tick() + if m := f.notifyMails(); len(m) != 0 { + t.Fatalf("a skipped change was mailed later: %+v", m) + } + }) + } +} + +func TestNotifierRestartAndNewWatchStaySilent(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + pat := f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.tick() + f.clk.Advance(25 * time.Hour) + f.tick() + if len(f.notifyMails()) != 1 { + t.Fatal("no mail for the change") + } + restarted := newNotifier(f.srv.auth, f.src, f.clk.Now) // the states live in users.db + f.setNode(evPkB, "Bravo", "companion", 48*time.Hour, nil) + if _, _, _, err := f.st.AddWatches(pat.me.ID, []string{evPkB}, 0); err != nil { + t.Fatal(err) + } + f.clk.Advance(5 * time.Minute) + restarted.tick(context.Background()) + if m := f.notifyMails(); len(m) != 1 { + t.Fatalf("the restart or the new watch mailed: %+v", m[1:]) + } +} + +func TestNotifierSendFailureIsNotRetried(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.tick() + f.fake.SetSendErr(errors.New("provider down")) + f.clk.Advance(25 * time.Hour) + f.tick() + f.fake.SetSendErr(nil) + f.clk.Advance(5 * time.Minute) + f.tick() + if m := f.notifyMails(); len(m) != 0 { + t.Fatalf("a failed send was retried: %+v", m) + } + if total, _, _ := f.st.NotifyMailCounts(f.clk.t.Add(-48 * time.Hour)); total != 0 { + t.Fatalf("mail_log has %d notify rows for a mail that never left", total) + } +} + +func TestNotifierStateWriteFailureSendsNothing(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.tick() + f.execDB(t, `CREATE TRIGGER notify_state_fail BEFORE UPDATE ON notification_state BEGIN SELECT RAISE(ABORT, 'boom'); END`) + f.clk.Advance(25 * time.Hour) + f.tick() + if m := f.notifyMails(); len(m) != 0 { + t.Fatalf("mailed although the states were not written: %+v", m) + } + f.execDB(t, `DROP TRIGGER notify_state_fail`) + f.tick() + if m := f.notifyMails(); len(m) != 1 { + t.Fatalf("mails after the write works again = %d; want 1", len(m)) + } +} + +func TestNotifierWaitsForTheAnalyzer(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.src.isReady = false + f.tick() + if states, _ := f.st.AllNotifyStates(); len(states) != 0 { + t.Fatalf("evaluated while the store was loading: %+v", states) + } + f.src.isReady = true + f.tick() + if states, _ := f.st.AllNotifyStates(); len(states) != 1 { + t.Fatalf("states once ready = %+v", states) + } +} + +func TestNotifierPanicIsRecovered(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.src.panicNext = true + f.n.safeTick(context.Background()) + f.tick() + if states, _ := f.st.AllNotifyStates(); len(states) != 1 { + t.Fatalf("the tick after the panic did not run: %+v", states) + } +} + +func TestNotifierReadsForeignAndObserversOnlyForAdmins(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.tick() + if f.src.foreignAsked != 0 || f.src.observersAsked != 0 { + t.Fatalf("admin data read without an admin asking: foreign %d, observers %d", f.src.foreignAsked, f.src.observersAsked) + } + boss := f.registerAndActivate(t, "admin@example.org", "Boss", pw) + if _, err := f.st.SetNotifyPrefs(boss.me.ID, true, []string{users.NotifyForeignNew}); err != nil { + t.Fatal(err) + } + f.tick() + if f.src.foreignAsked != 1 || f.src.observersAsked != 0 { + t.Fatalf("foreign %d, observers %d; want 1, 0", f.src.foreignAsked, f.src.observersAsked) + } + if err := f.st.SetRole(boss.me.ID, users.RoleUser); err != nil { + t.Fatal(err) + } + f.tick() + if f.src.foreignAsked != 1 { + t.Fatal("foreign nodes read for a demoted admin") + } +} + +func TestNotifyMailEscapesNodeNames(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, `Bad & "co"`, "companion", time.Hour, nil) + f.tick() + f.clk.Advance(25 * time.Hour) + f.tick() + m := f.notifyMails() + if len(m) != 1 || strings.Contains(m[0].HTML, "Bad") || !strings.Contains(m[0].HTML, "<b>Bad</b> & "co"") { + t.Fatalf("HTML = %q", m[0].HTML) + } +} + +// Names come from whoever owns a node or runs an observer: a line break +// must not forge a line in the mail, a bidi override must not reorder it. +func TestNotifyMailSanitisesNames(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA) + admin := f.registerAndActivate(t, "admin@example.org", "Ada", pw) + if _, err := f.st.SetNotifyPrefs(admin.me.ID, true, []string{users.NotifyObserverOffline}); err != nil { + t.Fatal(err) + } + f.setNode(evPkA, "x\r\nAccount suspended, sign in: https://evil.example\t\u202egnp.exe", "companion", time.Hour, nil) + seen := f.clk.t.Add(-time.Minute).Format(time.RFC3339) + f.src.obs = []notifyObserver{{ID: "OBS1", Name: "Roof\n\u2066Fake line\u2069", LastSeen: seen}, {ID: "OBS2", Name: "\n\u202e", LastSeen: seen}} + f.tick() + f.clk.Advance(25 * time.Hour) + f.tick() + m := f.notifyMails() + if len(m) != 2 { + t.Fatalf("mails = %d, want 2", len(m)) + } + for _, msg := range m { + for _, part := range []string{msg.Text, msg.HTML} { + for _, bad := range []string{"\r", "\t", "\u202e", "\u2066", "\u2069", "\nAccount", "\nFake"} { + if strings.Contains(part, bad) { + t.Fatalf("mail part contains %q:\n%s", bad, part) + } + } + } + } + body := m[0].Text + m[1].Text + for _, want := range []string{"- x Account suspended, sign in: https://evil.example gnp.exe: offline,", + "- Roof Fake line: observer offline,", "- OBS2: observer offline,"} { + if !strings.Contains(body, want) { + t.Fatalf("text lacks %q:\n%s", want, body) + } + } +} + +func TestMailSafeText(t *testing.T) { + cases := map[string]string{ + "plain": "plain", + " a \n\n b\r\tc ": "a b c", + "a\u202eb\u2066c\u2069d": "a b c d", + "a\u202ab\u202dc": "a b c", + "a\u0085b\x7fc": "a b c", + "\n\u202a": "", + "Zo\u00eb \u00e9": "Zo\u00eb \u00e9", + } + for in, want := range cases { + if got := mailSafeText(in); got != want { + t.Errorf("mailSafeText(%q) = %q; want %q", in, got, want) + } + } +} + +func TestNotifyChangeText(t *testing.T) { + at := time.Date(2026, 10, 7, 9, 5, 0, 0, time.UTC) + cases := []struct { + c notifyChange + want string + }{ + {notifyChange{Event: users.NotifyNodeOffline, Name: "A", To: users.NotifyBad, At: at}, "A: offline, 2026-10-07 09:05 UTC"}, + {notifyChange{Event: users.NotifyNodeOffline, Name: "A", To: users.NotifyGood, At: at}, "A: back online, 2026-10-07 09:05 UTC"}, + {notifyChange{Event: users.NotifyNodeBattery, Name: "A", To: users.NotifyBad, BatteryMv: evMv(3250), At: at}, "A: battery low (3250 mV), 2026-10-07 09:05 UTC"}, + {notifyChange{Event: users.NotifyNodeBattery, Name: "A", To: users.NotifyGood, BatteryMv: evMv(3410), At: at}, "A: battery recovered (3410 mV), 2026-10-07 09:05 UTC"}, + {notifyChange{Event: users.NotifyForeignNew, Name: "F", To: users.NotifyTold, At: at}, "F: new foreign node, 2026-10-07 09:05 UTC"}, + {notifyChange{Event: users.NotifyObserverOffline, Name: "O", To: users.NotifyBad, At: at}, "O: observer offline, 2026-10-07 09:05 UTC"}, + {notifyChange{Event: users.NotifyObserverOffline, Name: "O", To: users.NotifyGood, At: at}, "O: observer back online, 2026-10-07 09:05 UTC"}, + } + for _, c := range cases { + if got := notifyChangeText(c.c); got != c.want { + t.Errorf("%q; want %q", got, c.want) + } + } + if got := notifySubjectURL("https://e.org", notifyChange{Event: users.NotifyObserverOffline, Subject: "AB CD"}); got != "https://e.org/#/observers/AB%20CD" { + t.Errorf("observer URL = %q", got) + } + if got := notifySubjectURL("https://e.org", notifyChange{Event: users.NotifyNodeOffline, Subject: evPkA}); got != "https://e.org/#/nodes/"+evPkA { + t.Errorf("node URL = %q", got) + } +} + +func TestNotifySkipReason(t *testing.T) { + ns := defaultNotifySettings() + ok := users.User{Status: users.StatusActive} + on := users.NotifyPrefs{Enabled: true} + cases := []struct { + u users.User + p users.NotifyPrefs + mine, total int + want string + }{ + {ok, on, 0, 0, ""}, + {users.User{Status: users.StatusDisabled}, on, 0, 0, "account not active"}, + {users.User{Status: users.StatusActive, EmailBouncing: true}, on, 0, 0, "address bounces"}, + {ok, users.NotifyPrefs{}, 0, 0, "notifications off"}, + {ok, on, 20, 0, "daily limit for this user reached"}, + {ok, on, 19, 300, "daily limit for this instance reached"}, + {ok, on, 19, 299, ""}, + } + for _, c := range cases { + if got := notifySkipReason(c.u, c.p, c.mine, c.total, ns); got != c.want { + t.Errorf("%+v %+v %d %d = %q; want %q", c.u, c.p, c.mine, c.total, got, c.want) + } + } +} + +func TestInitUserManagementStartsTheNotifierOnlyWhenEnabled(t *testing.T) { + for _, on := range []bool{false, true} { + um := &UserManagementConfig{Enabled: true, PublicBaseURL: testBase, + Mail: UserMailConfig{BrevoAPIKey: "k", FromEmail: "noreply@example.org"}} + if on { + um.Notifications = &NotificationsConfig{Enabled: true} + } + srv := &Server{cfg: &Config{UserManagement: um}} + if err := srv.initUserManagement(filepath.Join(t.TempDir(), "meshcore.db")); err != nil { + t.Fatal(err) + } + if (srv.auth.notify != nil) != on { + t.Fatalf("notifications %v: notifier = %v", on, srv.auth.notify) + } + done := make(chan struct{}) + go func() { srv.closeUserManagement(); close(done) }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("closeUserManagement did not stop the notifier") + } + } +} + +func TestNotifierDeletesAdminStatesOfADemotedAdmin(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + boss := f.registerAndActivate(t, "admin@example.org", "Boss", pw) + if _, err := f.st.SetNotifyPrefs(boss.me.ID, true, []string{users.NotifyForeignNew}); err != nil { + t.Fatal(err) + } + f.src.nodeMap[evPkB] = notifyNode{Pubkey: evPkB, Name: "Far", Foreign: true} + f.tick() + if states, _ := f.st.AllNotifyStates(); len(states) != 2 { + t.Fatalf("baseline states = %+v; want the sentinel and Far", states) + } + if err := f.st.SetRole(boss.me.ID, users.RoleUser); err != nil { + t.Fatal(err) + } + f.tick() + if states, _ := f.st.AllNotifyStates(); len(states) != 0 { + t.Fatalf("states of a demoted admin = %+v; want none", states) + } + if err := f.st.SetRole(boss.me.ID, users.RoleAdmin); err != nil { + t.Fatal(err) + } + f.tick() + if m := f.notifyMails(); len(m) != 0 { + t.Fatalf("re-promotion mailed: %+v", m) + } +} + +func TestNotifierMailsAUserWithoutAPrefsRow(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + pat := f.registerAndActivate(t, "pat@example.org", "Pat", pw) + if _, _, _, err := f.st.AddWatches(pat.me.ID, []string{evPkA}, 0); err != nil { + t.Fatal(err) + } + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.tick() + if states, _ := f.st.AllNotifyStates(); len(states) != 1 { + t.Fatalf("states = %+v; want the default node.offline baseline", states) + } + if prefs, _ := f.st.AllNotifyPrefs(); len(prefs) != 0 { + t.Fatalf("a silent tick created prefs: %+v", prefs) + } + f.clk.Advance(25 * time.Hour) + f.tick() + m := f.notifyMails() + p, err := f.st.NotifyPrefsFor(pat.me.ID) + if err != nil { + t.Fatal(err) + } + if len(m) != 1 || p.UnsubToken == "" || !strings.Contains(m[0].Text, "unsubscribe?token="+p.UnsubToken) || + m[0].Headers["List-Unsubscribe"] != "<"+testBase+"/api/notifications/unsubscribe?token="+p.UnsubToken+">" { + t.Fatalf("mail = %+v, prefs = %+v", m, p) + } +} + +func TestNotifierNormalisesMixedCaseWatches(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + pat := f.watcher(t, "pat@example.org", "Pat", strings.ToUpper(evPkA)) + if _, _, _, err := f.st.AddWatches(pat.me.ID, []string{evPkA}, 0); err != nil { + t.Fatal(err) + } + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) // keyed lowercase, like NotifyNodes + f.src.heard[evPkA] = f.clk.t.Add(-time.Minute) + f.tick() + if len(f.src.heardAsked) != 1 || f.src.heardAsked[0] != evPkA { + t.Fatalf("last heard asked for %v; want [%s]", f.src.heardAsked, evPkA) + } + states, _ := f.st.AllNotifyStates() + if len(states) != 1 || states[0].Subject != evPkA || states[0].State != users.NotifyGood { + t.Fatalf("states = %+v; want one good row under the lowercase key", states) + } + delete(f.src.heard, evPkA) + f.clk.Advance(25 * time.Hour) + f.tick() + if m := f.notifyMails(); len(m) != 1 || !strings.Contains(m[0].Text, "Alpha: offline") || + strings.Count(m[0].Text, "Alpha:") != 1 { + t.Fatalf("mail = %+v; want one Alpha offline line", m) + } +} + +// cancelAfterSend cancels the tick's context once a mail went out, like a +// shutdown arriving mid-delivery. +type cancelAfterSend struct { + mailer.Mailer + cancel context.CancelFunc +} + +func (c cancelAfterSend) Send(ctx context.Context, m mailer.Message) (string, error) { + id, err := c.Mailer.Send(ctx, m) + c.cancel() + return id, err +} + +func TestNotifierStopsDeliveringAfterShutdown(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA) + f.watcher(t, "quinn@example.org", "Quinn", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.tick() + f.clk.Advance(25 * time.Hour) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + f.n.a.mail = cancelAfterSend{Mailer: f.n.a.mail, cancel: cancel} + f.n.tick(ctx) + if m := f.notifyMails(); len(m) != 1 { + t.Fatalf("mails after a shutdown mid-delivery = %d; want 1", len(m)) + } +} + +func TestNotifierPausesOfflineChecksWhileIngestIsStale(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.tick() // baseline: online + lastPacket := f.clk.t + f.src.newestAt = func() time.Time { return lastPacket } + + logs := captureLog(func() { + f.clk.Advance(25 * time.Hour) // feed down for a day: Alpha looks offline + f.tick() + f.clk.Advance(5 * time.Minute) + f.tick() + }) + if m := f.notifyMails(); len(m) != 0 { + t.Fatalf("mailed %d while ingest is stale", len(m)) + } + paused := "[notify] ingest stale since " + lastPacket.UTC().Format(time.RFC3339) + "; offline checks paused" + if n := strings.Count(logs, paused); n != 1 { + t.Fatalf("pause logged %d times; want once:\n%s", n, logs) + } + + f.src.newestAt = f.clk.Now // feed back, Alpha still silent + logs = captureLog(f.tick) + if m := f.notifyMails(); len(m) != 0 { + t.Fatalf("mailed %d on the first tick after recovery; offline waits one silent window", len(m)) + } + if n := strings.Count(logs, "[notify] ingest fresh again; offline checks resumed"); n != 1 { + t.Fatalf("resume logged %d times; want once:\n%s", n, logs) + } + f.clk.Advance(23 * time.Hour) + f.tick() + if m := f.notifyMails(); len(m) != 0 { + t.Fatalf("mailed %d inside the window after recovery", len(m)) + } + f.clk.Advance(time.Hour + time.Minute) // a full companion window after recovery + f.tick() + if m := f.notifyMails(); len(m) != 1 || !strings.Contains(m[0].Text, "Alpha: offline") { + t.Fatalf("mails one window after recovery = %+v; want Alpha offline", m) + } +} + +// A store that is already stale at startup gets the same grace once fresh. +func TestNotifierGraceWhenStaleAtStartup(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.tick() // baseline: online (states persist in users.db) + f.n = newNotifier(f.n.a, f.src, f.clk.Now) // restart + f.clk.Advance(25 * time.Hour) + old := f.clk.t.Add(-25 * time.Hour) + f.src.newestAt = func() time.Time { return old } + f.tick() // stale at startup + f.src.newestAt = f.clk.Now + f.tick() + if m := f.notifyMails(); len(m) != 0 { + t.Fatalf("mailed %d right after a stale startup recovered", len(m)) + } + f.clk.Advance(24*time.Hour + time.Minute) + f.tick() + if m := f.notifyMails(); len(m) != 1 { + t.Fatalf("mails one window after recovery = %d; want 1", len(m)) + } +} + +func TestNotifierTreatsAnEmptyStoreAsStale(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.src.newestAt = nil + if logs := captureLog(f.tick); !strings.Contains(logs, "[notify] ingest stale since (no packets); offline checks paused") { + t.Fatalf("log = %q", logs) + } + if s, err := f.st.AllNotifyStates(); err != nil || len(s) != 0 { + t.Fatalf("states with no packets in the store = %+v, %v", s, err) + } +} + +func TestNotifierLogsOneLinePerTick(t *testing.T) { + f := newNotifyFixture(t, defaultNotifySettings()) + f.watcher(t, "pat@example.org", "Pat", evPkA, evPkB) + f.watcher(t, "quinn@example.org", "Quinn", evPkB) + f.setNode(evPkA, "Alpha", "companion", time.Hour, nil) + f.setNode(evPkB, "Bravo", "companion", time.Hour, nil) + f.src.lock = 1500 * time.Microsecond + logs := captureLog(f.tick) + if !regexp.MustCompile(`\[notify\] tick: users=2 changes=0 mails=0 took=\d+\.\d\dms lock=1\.50ms`).MatchString(logs) { + t.Fatalf("first tick log = %q", logs) + } + f.clk.Advance(25 * time.Hour) + f.src.newestAt = f.clk.Now + logs = captureLog(f.tick) + if n := strings.Count(logs, "[notify] tick:"); n != 1 || !strings.Contains(logs, "users=2 changes=3 mails=2 ") { + t.Fatalf("second tick log (%d lines) = %q", n, logs) + } +} diff --git a/cmd/server/notify_source.go b/cmd/server/notify_source.go new file mode 100644 index 00000000..d069318a --- /dev/null +++ b/cmd/server/notify_source.go @@ -0,0 +1,205 @@ +package main + +import ( + "database/sql" + "strings" + "time" +) + +// notifySource is the analyzer side the notifier reads. serverNotifySource +// is the production one; tests substitute a fake. +type notifySource interface { + ready() bool // false while the packet store is still loading after startup + health() HealthThresholds + lowBatteryMv() int + nodes(pubkeys []string, withForeign bool) (map[string]notifyNode, error) + lastHeard(pubkeys []string) (map[string]time.Time, time.Duration) // and how long the store lock was held + lastRelayed(pubkeys []string) map[string]time.Time + observers() ([]notifyObserver, error) + newestPacket() time.Time // zero when the store holds no packet +} + +type serverNotifySource struct{ s *Server } + +// ready needs /api/healthz readiness (best-observation pick and neighbor +// graph done) and the finished startup load: hot window and background +// fill. Readiness alone can come while the newest packets are still +// loading. +func (src serverNotifySource) ready() bool { + if readiness.Load() == 0 || src.s.store == nil { + return false + } + select { + case <-src.s.store.StartupLoadDone(): + return true + default: + return false + } +} + +func (src serverNotifySource) health() HealthThresholds { return src.s.cfg.GetHealthThresholds() } +func (src serverNotifySource) lowBatteryMv() int { return src.s.cfg.LowBatteryMv() } + +func (src serverNotifySource) nodes(pubkeys []string, withForeign bool) (map[string]notifyNode, error) { + return src.s.db.NotifyNodes(pubkeys, withForeign) +} + +func (src serverNotifySource) lastHeard(pubkeys []string) (map[string]time.Time, time.Duration) { + if src.s.store == nil { + return map[string]time.Time{}, 0 + } + return src.s.store.LastHeardMap(pubkeys) +} + +// lastRelayed reads the cached relay metrics (the map /api/nodes uses; the +// background recomputer refreshes it). Pass repeaters and rooms only. +func (src serverNotifySource) lastRelayed(pubkeys []string) map[string]time.Time { + if src.s.store == nil || len(pubkeys) == 0 { + return map[string]time.Time{} + } + return relayTimes(src.s.store.GetRepeaterRelayInfoMap(src.s.cfg.GetHealthThresholds().RelayActiveHours), pubkeys) +} + +func (src serverNotifySource) newestPacket() time.Time { + if src.s.store == nil { + return time.Time{} + } + return src.s.store.NewestFirstSeen() +} + +func (src serverNotifySource) observers() ([]notifyObserver, error) { + list, err := src.s.db.GetObservers() + if err != nil { + return nil, err + } + return notifyObserversFrom(list), nil +} + +// relayTimes parses LastRelayed for the given keys; keys without a relay +// time are absent. +func relayTimes(m map[string]RepeaterRelayInfo, pubkeys []string) map[string]time.Time { + out := make(map[string]time.Time, len(pubkeys)) + for _, pk := range pubkeys { + info, ok := lookupRelayInfo(m, pk) + if !ok || info.LastRelayed == "" { + continue + } + if t, ok := parseRelayTS(info.LastRelayed); ok { + out[pk] = t + } + } + return out +} + +func notifyObserversFrom(list []Observer) []notifyObserver { + out := make([]notifyObserver, 0, len(list)) + for _, o := range list { + n := notifyObserver{ID: o.ID} + if o.Name != nil { + n.Name = *o.Name + } + if o.LastSeen != nil { + n.LastSeen = *o.LastSeen + } + out = append(out, n) + } + return out +} + +// notifyNodeChunk bounds the IN list of one query, far below SQLite's +// variable limit. +const notifyNodeChunk = 500 + +const notifyNodeCols = `SELECT public_key, COALESCE(name, ''), COALESCE(role, ''), COALESCE(last_seen, ''), battery_mv, COALESCE(foreign_advert, 0) FROM nodes` + +// NotifyNodes reads the node rows notifications need: the given pubkeys +// (primary-key lookups, in chunks) and, with withForeign, every node with +// foreign_advert = 1 (partial index idx_nodes_foreign_advert). Keyed by +// lowercase pubkey. Read-only, on the server's mode=ro handle. +func (db *DB) NotifyNodes(pubkeys []string, withForeign bool) (map[string]notifyNode, error) { + out := make(map[string]notifyNode, len(pubkeys)) + scan := func(rows *sql.Rows) error { + defer rows.Close() + for rows.Next() { + var n notifyNode + var mv sql.NullInt64 + var foreign int + if err := rows.Scan(&n.Pubkey, &n.Name, &n.Role, &n.LastSeen, &mv, &foreign); err != nil { + return err + } + n.Pubkey = strings.ToLower(n.Pubkey) + if mv.Valid { + v := int(mv.Int64) + n.BatteryMv = &v + } + n.Foreign = foreign != 0 + out[n.Pubkey] = n + } + return rows.Err() + } + for i := 0; i < len(pubkeys); i += notifyNodeChunk { + chunk := pubkeys[i:min(i+notifyNodeChunk, len(pubkeys))] + args := make([]any, len(chunk)) + for j, pk := range chunk { + args[j] = pk + } + rows, err := db.conn.Query(notifyNodeCols+` WHERE public_key IN (`+sqlPlaceholders(len(chunk))+`)`, args...) + if err != nil { + return nil, err + } + if err := scan(rows); err != nil { + return nil, err + } + } + if withForeign { + rows, err := db.conn.Query(notifyNodeCols + ` WHERE foreign_advert = 1`) + if err != nil { + return nil, err + } + if err := scan(rows); err != nil { + return nil, err + } + } + return out, nil +} + +// LastHeardMap returns, per pubkey with packets in the store, the newest +// FirstSeen among them: the node page's "Last Heard" (GetNodeHealth) in +// bulk, and how long the read lock was held (logged once per tick). One +// read lock, O(packets of the given nodes), string compares only; parsing +// happens after the lock is released. +func (s *PacketStore) LastHeardMap(pubkeys []string) (map[string]time.Time, time.Duration) { + latest := make(map[string]string, len(pubkeys)) + s.mu.RLock() + locked := time.Now() + for _, pk := range pubkeys { + for _, tx := range s.byNode[pk] { + if tx != nil && tx.FirstSeen > latest[pk] { + latest[pk] = tx.FirstSeen + } + } + } + held := time.Since(locked) + s.mu.RUnlock() + out := make(map[string]time.Time, len(latest)) + for pk, ts := range latest { + if t, ok := parseRelayTS(ts); ok { + out[pk] = t + } + } + return out, held +} + +// NewestFirstSeen is the FirstSeen of the newest transmission in the +// store, zero when there is none or it does not parse. O(1): s.packets is +// kept sorted by FirstSeen with the newest at the tail. +func (s *PacketStore) NewestFirstSeen() time.Time { + s.mu.RLock() + var ts string + if n := len(s.packets); n > 0 && s.packets[n-1] != nil { + ts = s.packets[n-1].FirstSeen + } + s.mu.RUnlock() + t, _ := parseRelayTS(ts) + return t +} diff --git a/cmd/server/notify_source_test.go b/cmd/server/notify_source_test.go new file mode 100644 index 00000000..1d5f025a --- /dev/null +++ b/cmd/server/notify_source_test.go @@ -0,0 +1,151 @@ +package main + +import ( + "fmt" + "testing" + "time" +) + +func TestDBNotifyNodes(t *testing.T) { + db := setupTestDB(t) + for _, q := range []string{ + fmt.Sprintf(`INSERT INTO nodes (public_key, name, role, last_seen, battery_mv, foreign_advert) VALUES ('%s', 'Alpha', 'repeater', '2026-10-07T11:00:00Z', 3712, 0)`, evPkA), + fmt.Sprintf(`INSERT INTO nodes (public_key, name, role, last_seen, foreign_advert) VALUES ('%s', NULL, 'companion', NULL, 0)`, evPkB), + fmt.Sprintf(`INSERT INTO nodes (public_key, name, role, last_seen, foreign_advert) VALUES ('%s', 'Foxtrot', 'repeater', '2026-10-07T10:00:00Z', 1)`, evPkF), + } { + if _, err := db.conn.Exec(q); err != nil { + t.Fatal(err) + } + } + got, err := db.NotifyNodes([]string{evPkA, evPkB, evPkG}, false) + if err != nil { + t.Fatal(err) + } + a, b := got[evPkA], got[evPkB] + if len(got) != 2 || a.Pubkey != evPkA || a.Name != "Alpha" || a.Role != "repeater" || a.LastSeen != "2026-10-07T11:00:00Z" || + a.BatteryMv == nil || *a.BatteryMv != 3712 || a.Foreign { + t.Fatalf("NotifyNodes = %+v", got) + } + if b.Name != "" || b.LastSeen != "" || b.BatteryMv != nil { + t.Fatalf("NULL columns = %+v", b) + } + got, err = db.NotifyNodes([]string{evPkA}, true) + if err != nil || len(got) != 2 || !got[evPkF].Foreign || got[evPkF].Name != "Foxtrot" { + t.Fatalf("with foreign = %+v, %v", got, err) + } + if got, err := db.NotifyNodes(nil, false); err != nil || len(got) != 0 { + t.Fatalf("no keys = %+v, %v", got, err) + } +} + +func TestDBNotifyNodesChunksLongLists(t *testing.T) { + db := setupTestDB(t) + keys := make([]string, 0, 1200) + for i := 0; i < 1200; i++ { + keys = append(keys, fmt.Sprintf("%064x", i)) + } + for _, i := range []int{3, 600, 1199} { + if _, err := db.conn.Exec(`INSERT INTO nodes (public_key, role) VALUES (?, 'companion')`, keys[i]); err != nil { + t.Fatal(err) + } + } + got, err := db.NotifyNodes(keys, false) + if err != nil || len(got) != 3 { + t.Fatalf("1200 keys over %d-key chunks = %d nodes, %v; want 3", notifyNodeChunk, len(got), err) + } +} + +func TestPacketStoreLastHeardMap(t *testing.T) { + s := &PacketStore{byNode: map[string][]*StoreTx{ + evPkA: {{FirstSeen: "2026-10-07T10:00:00Z"}, {FirstSeen: "2026-10-07T11:30:00.000Z"}, nil, {FirstSeen: "2026-10-07T09:00:00Z"}}, + evPkB: {{FirstSeen: "not a time"}}, + }} + got, lock := s.LastHeardMap([]string{evPkA, evPkB, evPkF}) + if len(got) != 1 || !got[evPkA].Equal(time.Date(2026, 10, 7, 11, 30, 0, 0, time.UTC)) { + t.Fatalf("LastHeardMap = %v", got) + } + if lock < 0 || lock > time.Second { + t.Fatalf("lock time = %v", lock) + } +} + +// 5,000 watched nodes with 20 packets each, the spec's upper figure. Quote +// the result in the PR: this runs under the store's read lock. +func BenchmarkLastHeardMap(b *testing.B) { + s := &PacketStore{byNode: map[string][]*StoreTx{}} + keys := make([]string, 0, 5000) + for i := 0; i < 5000; i++ { + pk := fmt.Sprintf("%064x", i) + keys = append(keys, pk) + for j := 0; j < 20; j++ { + s.byNode[pk] = append(s.byNode[pk], &StoreTx{FirstSeen: evNow.Add(-time.Duration(j) * time.Minute).Format(time.RFC3339)}) + } + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + s.LastHeardMap(keys) + } +} + +func TestRelayTimes(t *testing.T) { + m := map[string]RepeaterRelayInfo{ + evPkA: {LastRelayed: "2026-10-07T11:00:00Z"}, + evPkB: {}, + } + got := relayTimes(m, []string{evPkA, evPkB, evPkF}) + if len(got) != 1 || !got[evPkA].Equal(time.Date(2026, 10, 7, 11, 0, 0, 0, time.UTC)) { + t.Fatalf("relayTimes = %v", got) + } +} + +func TestNotifyObserversFrom(t *testing.T) { + name, seen := "Roof", "2026-10-07T11:00:00Z" + got := notifyObserversFrom([]Observer{{ID: "OBS1", Name: &name, LastSeen: &seen}, {ID: "OBS2"}}) + if len(got) != 2 || got[0] != (notifyObserver{ID: "OBS1", Name: "Roof", LastSeen: seen}) || got[1] != (notifyObserver{ID: "OBS2"}) { + t.Fatalf("notifyObserversFrom = %+v", got) + } +} + +func TestServerNotifySourceWithoutStore(t *testing.T) { + src := serverNotifySource{s: &Server{cfg: &Config{}}} + if heard, lock := src.lastHeard([]string{evPkA}); len(heard) != 0 || lock != 0 || len(src.lastRelayed([]string{evPkA})) != 0 { + t.Fatal("a server without a packet store reported times") + } + if src.lowBatteryMv() != 3300 || src.health().InfraSilentHours != 72 { + t.Fatal("defaults not taken from the config") + } +} + +// The startup load (hot window plus background fill) must be finished, not +// only /api/healthz readiness: readiness can be reached while the newest +// packets are still loading, which would mail "offline" then "back online". +func TestServerNotifySourceReadyWaitsForTheStartupLoad(t *testing.T) { + readiness.Store(1) + defer readiness.Store(0) + store := &PacketStore{clockSkew: &ClockSkewEngine{}} + src := serverNotifySource{s: &Server{cfg: &Config{}, store: store}} + if src.ready() { + t.Fatal("ready before the startup load finished") + } + store.signalStartupLoadDone() + if !src.ready() { + t.Fatal("not ready after readiness and the startup load") + } + readiness.Store(0) + if src.ready() { + t.Fatal("ready without readiness") + } + if (serverNotifySource{s: &Server{cfg: &Config{}}}).ready() { + t.Fatal("ready without a packet store") + } +} + +func TestPacketStoreNewestFirstSeen(t *testing.T) { + if got := (&PacketStore{}).NewestFirstSeen(); !got.IsZero() { + t.Fatalf("empty store = %v", got) + } + s := &PacketStore{packets: []*StoreTx{{FirstSeen: "2026-10-07T10:00:00Z"}, {FirstSeen: "2026-10-07T11:30:00.000Z"}}} + if got := s.NewestFirstSeen(); !got.Equal(time.Date(2026, 10, 7, 11, 30, 0, 0, time.UTC)) { + t.Fatalf("NewestFirstSeen = %v", got) + } +} diff --git a/cmd/server/openapi.go b/cmd/server/openapi.go index f7fd68f5..b68255a8 100644 --- a/cmd/server/openapi.go +++ b/cmd/server/openapi.go @@ -95,6 +95,15 @@ func routeDescriptions() map[string]routeMeta { "POST /api/admin/proposals/{id}/revoke": {Summary: "Revoke an approved proposal (admin)", Description: "Body {note}. Approved only (409 otherwise). The ingestor stops decrypting the channel from its next refresh unless the channel is configured; stored messages stay. The name can be proposed again.", Tag: "users", Session: true}, "POST /api/mail/brevo/webhook": {Summary: "Brevo delivery-event webhook", Description: "Authenticated with Authorization: Bearer