Add notification channels and a provider abstraction
Users get a notification_channels table (replacing users.ntfy_topic); pages fan
out to every enabled channel through a notify.Provider registry, with ntfy as
the first provider. Fallbacks are typed {type, address}. Schema edited in place:
existing databases must be recreated.
Claude-Session: https://claude.ai/code/session_016mBLURvJoMuUEr9cB2RpUN
This commit is contained in:
+217
-209
@@ -1,18 +1,16 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.ryuvia.com/niklas/terdut-server/internal/models"
|
||||
"git.ryuvia.com/niklas/terdut-server/internal/notify"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -20,8 +18,8 @@ const (
|
||||
// 15 minute tick is far too coarse for something that has to wake a person.
|
||||
notifyInterval = 30 * time.Second
|
||||
|
||||
// notifyRetryBase and notifyRetryMax bound the delivery backoff. ntfy being
|
||||
// briefly unreachable should not lose the page.
|
||||
// notifyRetryBase and notifyRetryMax bound the delivery backoff. A provider
|
||||
// being briefly unreachable should not lose the page.
|
||||
notifyRetryBase = 30 * time.Second
|
||||
notifyRetryMax = 15 * time.Minute
|
||||
|
||||
@@ -48,43 +46,41 @@ const notifierLockKey int64 = 7265_0002
|
||||
|
||||
// Notification kinds, recording why a push was sent.
|
||||
const (
|
||||
notifyTriggered = "triggered"
|
||||
notifyReminder = "reminder"
|
||||
notifyResolved = "resolved"
|
||||
notifyTriggered = notify.KindTriggered
|
||||
notifyReminder = notify.KindReminder
|
||||
notifyResolved = notify.KindResolved
|
||||
|
||||
// notifyEscalated is a page that went out because nobody answered the last
|
||||
// one. Told apart from a reminder because it goes to somebody else.
|
||||
notifyEscalated = "escalated"
|
||||
notifyEscalated = notify.KindEscalated
|
||||
)
|
||||
|
||||
// Timeline event types the notifier writes, so an incident's history says who
|
||||
// was paged and whether the page landed. Written from the delivery result
|
||||
// rather than at enqueue: a queued notification is an intention, and claiming
|
||||
// somebody was told before ntfy accepted it would be a lie the timeline keeps.
|
||||
// somebody was told before the provider accepted it would be a lie the timeline
|
||||
// keeps.
|
||||
//
|
||||
// The topic is deliberately absent from both. It is a shared secret with the
|
||||
// ntfy server — anyone holding it can publish to it — and the timeline is
|
||||
// readable by every API key.
|
||||
// The channel's address is deliberately absent from both; only its type is
|
||||
// named. An address can be a shared secret (an ntfy topic lets anyone holding it
|
||||
// publish to it) and the timeline is readable by every API key.
|
||||
const (
|
||||
eventNotified = "notified"
|
||||
eventNotifyFailed = "notify_failed"
|
||||
)
|
||||
|
||||
// NotifyConfig is everything the notifier needs to reach ntfy and to build URLs
|
||||
// a phone can follow back to this server.
|
||||
// NotifyConfig is everything the notifier needs to reach people and to build
|
||||
// URLs they can follow back to this server.
|
||||
type NotifyConfig struct {
|
||||
// BaseURL is the ntfy server. Empty disables notifications entirely: no
|
||||
// goroutine, and nothing is ever enqueued.
|
||||
BaseURL string
|
||||
// Providers are the ways this server can deliver a notification. Empty
|
||||
// disables notifications entirely: no goroutine, and nothing is ever enqueued.
|
||||
Providers *notify.Registry
|
||||
|
||||
// Token is an optional bearer token for an access-controlled ntfy.
|
||||
Token string
|
||||
|
||||
// FallbackTopic receives incidents that open with nobody on call. Those
|
||||
// notifications carry no Acknowledge button — there is no user to attribute
|
||||
// the acknowledgement to, and putting one on a shared topic would let any
|
||||
// subscriber acknowledge as somebody else.
|
||||
FallbackTopic string
|
||||
// Fallback receives incidents that open with nobody on call who can be
|
||||
// reached. Those notifications carry no Acknowledge button — there is no user
|
||||
// to attribute the acknowledgement to, and putting one on a shared target
|
||||
// would let any recipient acknowledge as somebody else.
|
||||
Fallback notify.Target
|
||||
|
||||
// PublicURL is the base URL a phone uses to reach this server, for the
|
||||
// notification's click target and its Acknowledge action. Without it a
|
||||
@@ -97,11 +93,7 @@ type NotifyConfig struct {
|
||||
}
|
||||
|
||||
// enabled reports whether notifications are configured at all.
|
||||
func (c NotifyConfig) enabled() bool { return c.BaseURL != "" }
|
||||
|
||||
// notifyClient is shared: a page is small and infrequent, and the timeout is
|
||||
// what keeps a hung ntfy from stalling the delivery pass.
|
||||
var notifyClient = &http.Client{Timeout: 10 * time.Second}
|
||||
func (c NotifyConfig) enabled() bool { return !c.Providers.Empty() }
|
||||
|
||||
// StartNotifier delivers queued notifications until ctx is cancelled, starting
|
||||
// with an immediate pass so a restart flushes whatever the last one left behind.
|
||||
@@ -111,10 +103,14 @@ var notifyClient = &http.Client{Timeout: 10 * time.Second}
|
||||
// delivers; the rest skip that tick rather than racing the same pass.
|
||||
func StartNotifier(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||
if !cfg.enabled() {
|
||||
log.Print("notifier: disabled (no ntfy URL configured)")
|
||||
log.Print("notifier: disabled (no notification provider configured)")
|
||||
return
|
||||
}
|
||||
log.Printf("notifier: publishing to %s", cfg.BaseURL)
|
||||
var types []string
|
||||
for _, p := range cfg.Providers.List() {
|
||||
types = append(types, p.Type())
|
||||
}
|
||||
log.Printf("notifier: delivering through %s", strings.Join(types, ", "))
|
||||
|
||||
ticker := time.NewTicker(notifyInterval)
|
||||
defer ticker.Stop()
|
||||
@@ -156,8 +152,13 @@ func NotifySweep(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||
// why there is no separate reminder cap.
|
||||
//
|
||||
// The previous notification must have actually been sent before another is
|
||||
// queued, so an ntfy outage produces a retry backlog rather than a reminder
|
||||
// backlog that all lands at once when it comes back.
|
||||
// queued, so a provider outage produces a retry backlog rather than a reminder
|
||||
// backlog that all lands at once when it comes back. "Previous notification"
|
||||
// is the incident's latest batch: everything one fan-out queued shares a
|
||||
// created_at, and the batch counts as sent once any of its rows has landed.
|
||||
//
|
||||
// A channel the user has since disabled or removed is not reminded: that is the
|
||||
// off switch for a channel, and a reminder that ignored it would defeat it.
|
||||
func enqueueReminders(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||
// cfg.RepeatEvery is what the server started with; the settings table is
|
||||
// what it runs on. Read per tick, so an administrator lengthening the
|
||||
@@ -168,28 +169,30 @@ func enqueueReminders(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||
}
|
||||
now := time.Now()
|
||||
|
||||
type due struct {
|
||||
incidentID int64
|
||||
userID *int64
|
||||
topic string
|
||||
}
|
||||
|
||||
rows, err := db.QueryContext(ctx, `
|
||||
SELECT n.incident_id, n.user_id, n.topic
|
||||
SELECT n.incident_id, n.user_id, n.channel_id, n.channel_type, n.address
|
||||
FROM notifications n
|
||||
JOIN incidents i ON i.id = n.incident_id
|
||||
WHERE n.id = (SELECT MAX(id) FROM notifications WHERE incident_id = n.incident_id)
|
||||
AND n.sent_at IS NOT NULL
|
||||
LEFT JOIN notification_channels c ON c.id = n.channel_id
|
||||
WHERE n.created_at = (SELECT MAX(created_at) FROM notifications WHERE incident_id = n.incident_id)
|
||||
AND EXISTS (SELECT 1 FROM notifications s
|
||||
WHERE s.incident_id = n.incident_id
|
||||
AND s.created_at = n.created_at
|
||||
AND s.sent_at IS NOT NULL)
|
||||
AND n.created_at <= $1
|
||||
AND i.resolved_at IS NULL
|
||||
AND i.archived_at IS NULL
|
||||
AND i.status = 'triggered'
|
||||
AND (i.snoozed_until IS NULL OR i.snoozed_until <= $2)
|
||||
-- A shared fallback target has no user and no channel row; everything
|
||||
-- else must still point at an enabled channel.
|
||||
AND (n.user_id IS NULL OR c.enabled)
|
||||
-- A team with an escalation ladder gets escalation instead. Both
|
||||
-- would mean two pages for one silence, which is how people learn to
|
||||
-- mute a tool.
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM escalation_levels el WHERE el.team_id = i.team_id)`,
|
||||
SELECT 1 FROM escalation_levels el WHERE el.team_id = i.team_id)
|
||||
ORDER BY n.id`,
|
||||
now.Add(-repeat).Unix(), now.Unix())
|
||||
if err != nil {
|
||||
log.Printf("notifier: find reminders: %v", err)
|
||||
@@ -199,15 +202,20 @@ func enqueueReminders(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||
// Collected before inserting, rather than written while walking the cursor:
|
||||
// the inserts below are what this query selects on, and a cursor reading its
|
||||
// own writes is a hazard whatever the pool size.
|
||||
var pending []due
|
||||
byIncident := map[int64][]recipient{}
|
||||
var order []int64
|
||||
for rows.Next() {
|
||||
var d due
|
||||
if err := rows.Scan(&d.incidentID, &d.userID, &d.topic); err != nil {
|
||||
var incidentID int64
|
||||
var r recipient
|
||||
if err := rows.Scan(&incidentID, &r.userID, &r.channelID, &r.typ, &r.address); err != nil {
|
||||
rows.Close()
|
||||
log.Printf("notifier: scan reminder: %v", err)
|
||||
return
|
||||
}
|
||||
pending = append(pending, d)
|
||||
if _, seen := byIncident[incidentID]; !seen {
|
||||
order = append(order, incidentID)
|
||||
}
|
||||
byIncident[incidentID] = append(byIncident[incidentID], r)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
rows.Close()
|
||||
@@ -216,13 +224,15 @@ func enqueueReminders(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||
}
|
||||
rows.Close()
|
||||
|
||||
for _, d := range pending {
|
||||
if err := enqueueNotification(ctx, db, d.incidentID, d.userID, d.topic, notifyReminder); err != nil {
|
||||
log.Printf("notifier: queue reminder for incident %d: %v", d.incidentID, err)
|
||||
// One batch per incident, so the reminder is itself a batch the next one
|
||||
// can be measured from.
|
||||
for _, id := range order {
|
||||
if err := enqueueNotifications(ctx, db, id, byIncident[id], notifyReminder); err != nil {
|
||||
log.Printf("notifier: queue reminder for incident %d: %v", id, err)
|
||||
}
|
||||
}
|
||||
if len(pending) > 0 {
|
||||
log.Printf("notifier: queued %d reminder(s)", len(pending))
|
||||
if len(order) > 0 {
|
||||
log.Printf("notifier: queued reminders for %d incident(s)", len(order))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -231,7 +241,8 @@ type outboxRow struct {
|
||||
id int64
|
||||
incidentID int64
|
||||
userID *int64
|
||||
topic string
|
||||
channel string // channel type; the provider that delivers it
|
||||
address string
|
||||
kind string
|
||||
attempts int
|
||||
}
|
||||
@@ -258,7 +269,8 @@ func deliverPending(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||
}
|
||||
// Logged, not returned: the page has already gone out, and treating a
|
||||
// failed timeline write as a failed delivery would send it again.
|
||||
if err := logEvent(ctx, db, n.incidentID, eventNotified, n.userID, nil, nil, &n.kind); err != nil {
|
||||
detail := deliveryDetail(n)
|
||||
if err := logEvent(ctx, db, n.incidentID, eventNotified, n.userID, nil, nil, &detail); err != nil {
|
||||
log.Printf("notifier: log delivery of %d: %v", n.id, err)
|
||||
}
|
||||
sent++
|
||||
@@ -272,7 +284,7 @@ func deliverPending(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||
// caller writes, for the same single-connection reason as staleAlertIDs.
|
||||
func pendingNotifications(ctx context.Context, db *sql.DB) ([]outboxRow, error) {
|
||||
rows, err := db.QueryContext(ctx, `
|
||||
SELECT id, incident_id, user_id, topic, kind, attempts
|
||||
SELECT id, incident_id, user_id, channel_type, address, kind, attempts
|
||||
FROM notifications
|
||||
WHERE sent_at IS NULL
|
||||
AND send_after <= $1
|
||||
@@ -287,7 +299,7 @@ func pendingNotifications(ctx context.Context, db *sql.DB) ([]outboxRow, error)
|
||||
var batch []outboxRow
|
||||
for rows.Next() {
|
||||
var n outboxRow
|
||||
if err := rows.Scan(&n.id, &n.incidentID, &n.userID, &n.topic, &n.kind, &n.attempts); err != nil {
|
||||
if err := rows.Scan(&n.id, &n.incidentID, &n.userID, &n.channel, &n.address, &n.kind, &n.attempts); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
batch = append(batch, n)
|
||||
@@ -312,12 +324,18 @@ func markFailed(ctx context.Context, db *sql.DB, n outboxRow, cause error) {
|
||||
if n.attempts+1 < notifyMaxAttempts {
|
||||
return
|
||||
}
|
||||
detail := fmt.Sprintf("%s: %s", n.kind, cause)
|
||||
detail := fmt.Sprintf("%s: %s", deliveryDetail(n), cause)
|
||||
if err := logEvent(ctx, db, n.incidentID, eventNotifyFailed, n.userID, nil, nil, &detail); err != nil {
|
||||
log.Printf("notifier: log failure of %d: %v", n.id, err)
|
||||
}
|
||||
}
|
||||
|
||||
// deliveryDetail is what a timeline event says about a delivery: the kind and
|
||||
// the channel type, "triggered via ntfy". Never the address.
|
||||
func deliveryDetail(n outboxRow) string {
|
||||
return n.kind + " via " + n.channel
|
||||
}
|
||||
|
||||
// retryDelay doubles the wait per attempt, up to notifyRetryMax.
|
||||
func retryDelay(attempts int) time.Duration {
|
||||
d := notifyRetryBase << attempts
|
||||
@@ -328,10 +346,17 @@ func retryDelay(attempts int) time.Duration {
|
||||
}
|
||||
|
||||
// deliver renders one notification against the incident's *current* state and
|
||||
// publishes it. Rendering happens here rather than at enqueue time so a message
|
||||
// that waited in the queue while its incident escalated goes out at the
|
||||
// severity the incident has now.
|
||||
// hands it to the provider its channel names. Rendering happens here rather than
|
||||
// at enqueue time so a message that waited in the queue while its incident
|
||||
// escalated goes out at the severity the incident has now.
|
||||
func deliver(ctx context.Context, db *sql.DB, cfg NotifyConfig, n outboxRow) error {
|
||||
// Looked up first: a row for a provider this server no longer has can never
|
||||
// succeed, and there is no point rendering it.
|
||||
provider, ok := cfg.Providers.Get(n.channel)
|
||||
if !ok {
|
||||
return fmt.Errorf("no %q provider is configured", n.channel)
|
||||
}
|
||||
|
||||
inc, err := fetchIncident(ctx, db, n.incidentID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("load incident: %w", err)
|
||||
@@ -354,59 +379,43 @@ func deliver(ctx context.Context, db *sql.DB, cfg NotifyConfig, n outboxRow) err
|
||||
if n.kind == notifyTriggered {
|
||||
if sim, err := similarIncidents(ctx, db, n.incidentID, 1); err == nil && len(sim) > 0 && len(sim[0].ResolutionNotes) > 0 {
|
||||
notes := sim[0].ResolutionNotes
|
||||
msg.Message += "\nLast time: " + shorten(derefString(notes[len(notes)-1].Detail), 160)
|
||||
msg.Body += "\nLast time: " + shorten(derefString(notes[len(notes)-1].Detail), 160)
|
||||
}
|
||||
}
|
||||
|
||||
// An Acknowledge button needs both a user to attribute the acknowledgement
|
||||
// to and a URL the phone can reach. Minted per delivery, so every push
|
||||
// carries its own short-lived token rather than reusing one.
|
||||
// An Acknowledge control needs both a user to attribute the acknowledgement
|
||||
// to and a URL the recipient can reach.
|
||||
if n.kind != notifyResolved && n.userID != nil && cfg.PublicURL != "" {
|
||||
raw, err := issueAckToken(ctx, db, n.incidentID, *n.userID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("issue ack token: %w", err)
|
||||
base := strings.TrimSuffix(cfg.PublicURL, "/")
|
||||
if provider.Capabilities().ActionButtons {
|
||||
// Minted per delivery, so every push carries its own short-lived
|
||||
// token rather than reusing one.
|
||||
raw, err := issueAckToken(ctx, db, n.incidentID, *n.userID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("issue ack token: %w", err)
|
||||
}
|
||||
msg.AckURL = base + "/api/notify/ack/" + raw
|
||||
} else {
|
||||
// No button to put it on, so the way to acknowledge travels as text.
|
||||
// The token URL is POST-only and a link to it would do nothing, and
|
||||
// minting one nobody can use is waste: this is the incident page,
|
||||
// which has the Acknowledge control.
|
||||
msg.Body += fmt.Sprintf("\nAcknowledge: %s/incidents/%d", base, n.incidentID)
|
||||
}
|
||||
msg.Actions = append(msg.Actions, ntfyAction{
|
||||
Action: "http",
|
||||
Label: "Acknowledge",
|
||||
URL: strings.TrimSuffix(cfg.PublicURL, "/") + "/api/notify/ack/" + raw,
|
||||
Method: "POST",
|
||||
Clear: true,
|
||||
})
|
||||
}
|
||||
|
||||
return publish(ctx, cfg, msg)
|
||||
return provider.Send(ctx, msg)
|
||||
}
|
||||
|
||||
// ntfyMessage is ntfy's JSON publish format. Using it rather than the X-Actions
|
||||
// header avoids that header's comma and quote escaping rules, which are easy to
|
||||
// break with a title that happens to contain a comma.
|
||||
type ntfyMessage struct {
|
||||
Topic string `json:"topic"`
|
||||
Title string `json:"title,omitempty"`
|
||||
Message string `json:"message"`
|
||||
Priority int `json:"priority,omitempty"`
|
||||
Tags []string `json:"tags,omitempty"`
|
||||
Click string `json:"click,omitempty"`
|
||||
Actions []ntfyAction `json:"actions,omitempty"`
|
||||
}
|
||||
|
||||
type ntfyAction struct {
|
||||
Action string `json:"action"`
|
||||
Label string `json:"label"`
|
||||
URL string `json:"url"`
|
||||
Method string `json:"method,omitempty"`
|
||||
Clear bool `json:"clear,omitempty"`
|
||||
}
|
||||
|
||||
// renderNotification builds the message body for one queued notification.
|
||||
func renderNotification(inc models.Incident, n outboxRow, firing int, cfg NotifyConfig) ntfyMessage {
|
||||
msg := ntfyMessage{Topic: n.topic}
|
||||
// renderNotification builds the provider-neutral message for one queued
|
||||
// notification.
|
||||
func renderNotification(inc models.Incident, n outboxRow, firing int, cfg NotifyConfig) notify.Message {
|
||||
msg := notify.Message{Address: n.address, Kind: n.kind}
|
||||
|
||||
if cfg.PublicURL != "" {
|
||||
// The web UI's page for the incident, so tapping the notification
|
||||
// opens something a browser can use.
|
||||
msg.Click = fmt.Sprintf("%s/incidents/%d",
|
||||
msg.ClickURL = fmt.Sprintf("%s/incidents/%d",
|
||||
strings.TrimSuffix(cfg.PublicURL, "/"), inc.ID)
|
||||
}
|
||||
|
||||
@@ -415,10 +424,8 @@ func renderNotification(inc models.Incident, n outboxRow, firing int, cfg Notify
|
||||
switch n.kind {
|
||||
case notifyResolved:
|
||||
msg.Title = "Resolved: " + title
|
||||
msg.Message = "All alerts stopped firing after " +
|
||||
msg.Body = "All alerts stopped firing after " +
|
||||
humanDuration(time.Since(inc.TriggeredAt))
|
||||
msg.Priority = ntfyPriorityLow
|
||||
msg.Tags = []string{"white_check_mark"}
|
||||
return msg
|
||||
|
||||
case notifyReminder:
|
||||
@@ -440,9 +447,8 @@ func renderNotification(inc models.Incident, n outboxRow, firing int, cfg Notify
|
||||
parts = append(parts, "open "+humanDuration(time.Since(inc.TriggeredAt)))
|
||||
}
|
||||
|
||||
msg.Message = strings.Join(parts, " · ")
|
||||
msg.Priority = ntfyPriority(severity)
|
||||
msg.Tags = []string{severityTag(severity)}
|
||||
msg.Body = strings.Join(parts, " · ")
|
||||
msg.Level = severityRank(severity)
|
||||
return msg
|
||||
}
|
||||
|
||||
@@ -488,137 +494,139 @@ func titleWithoutLabel(title, key, value string) string {
|
||||
return m[1] + " (" + strings.Join(rest, ", ") + ")"
|
||||
}
|
||||
|
||||
// ntfy's priority scale. Max is the one that overrides the phone's quiet
|
||||
// settings, which is the whole point of paging on critical.
|
||||
const (
|
||||
ntfyPriorityLow = 2
|
||||
ntfyPriorityDefault = 3
|
||||
ntfyPriorityHigh = 4
|
||||
ntfyPriorityMax = 5
|
||||
)
|
||||
|
||||
// ntfyPriority maps an incident's severity onto ntfy's scale, following the
|
||||
// same ordering severityRank uses. An unrecognised severity gets the default
|
||||
// rather than being silenced.
|
||||
func ntfyPriority(severity string) int {
|
||||
switch severityRank(severity) {
|
||||
case 4:
|
||||
return ntfyPriorityMax
|
||||
case 3:
|
||||
return ntfyPriorityHigh
|
||||
case 2:
|
||||
return ntfyPriorityDefault
|
||||
case 1:
|
||||
return ntfyPriorityLow
|
||||
default:
|
||||
return ntfyPriorityDefault
|
||||
}
|
||||
// recipient is one place a notification goes: a user's channel, or the shared
|
||||
// fallback target. channelID and userID are nil for the fallback, which belongs
|
||||
// to nobody. A batch may hold several users (an escalation level with more than
|
||||
// one target), which is why the user is here and not on the batch.
|
||||
type recipient struct {
|
||||
userID *int64
|
||||
channelID *int64
|
||||
typ string
|
||||
address string
|
||||
}
|
||||
|
||||
func severityTag(severity string) string {
|
||||
switch severityRank(severity) {
|
||||
case 4:
|
||||
return "rotating_light"
|
||||
case 3:
|
||||
return "red_circle"
|
||||
case 2:
|
||||
return "warning"
|
||||
case 1:
|
||||
return "information_source"
|
||||
default:
|
||||
return "bell"
|
||||
}
|
||||
}
|
||||
|
||||
// publish POSTs one message to ntfy.
|
||||
func publish(ctx context.Context, cfg NotifyConfig, msg ntfyMessage) error {
|
||||
body, err := json.Marshal(msg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
|
||||
strings.TrimSuffix(cfg.BaseURL, "/"), bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if cfg.Token != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+cfg.Token)
|
||||
}
|
||||
|
||||
resp, err := notifyClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return fmt.Errorf("ntfy returned %s", resp.Status)
|
||||
// enqueueNotifications queues one row per recipient for a single notification.
|
||||
//
|
||||
// The rows share one created_at. That is what makes them a batch: the reminder
|
||||
// and all-clear paths find "who was told last" as the rows carrying an
|
||||
// incident's newest created_at, so a fan-out to three channels is reminded and
|
||||
// resolved as one unit rather than as whichever row happened to land last.
|
||||
func enqueueNotifications(ctx context.Context, q querier, incidentID int64, recipients []recipient, kind string) error {
|
||||
now := time.Now().Unix()
|
||||
for _, r := range recipients {
|
||||
if _, err := q.ExecContext(ctx, `
|
||||
INSERT INTO notifications (incident_id, user_id, channel_id, channel_type, address, kind, created_at, send_after)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $7)`,
|
||||
incidentID, r.userID, r.channelID, r.typ, r.address, kind, now); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// enqueueNotification adds one row to the outbox, due immediately.
|
||||
func enqueueNotification(ctx context.Context, q querier, incidentID int64, userID *int64, topic, kind string) error {
|
||||
now := time.Now().Unix()
|
||||
_, err := q.ExecContext(ctx, `
|
||||
INSERT INTO notifications (incident_id, user_id, topic, kind, created_at, send_after)
|
||||
VALUES ($1, $2, $3, $4, $5, $6)`, incidentID, userID, topic, kind, now, now)
|
||||
return err
|
||||
// userRecipients is the channels a notification to this user should go to: the
|
||||
// ones they have enabled, of a type this server can deliver. A channel whose
|
||||
// provider is not configured is skipped rather than queued, since it could only
|
||||
// fail eight times and then sit in the table.
|
||||
func userRecipients(ctx context.Context, q querier, cfg NotifyConfig, userID int64) ([]recipient, error) {
|
||||
uid := userID
|
||||
rows, err := q.QueryContext(ctx, `
|
||||
SELECT id, type, address FROM notification_channels
|
||||
WHERE user_id = $1 AND enabled
|
||||
ORDER BY id`, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []recipient
|
||||
for rows.Next() {
|
||||
var id int64
|
||||
var r recipient
|
||||
if err := rows.Scan(&id, &r.typ, &r.address); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !cfg.Providers.Has(r.typ) {
|
||||
continue
|
||||
}
|
||||
r.channelID, r.userID = &id, &uid
|
||||
out = append(out, r)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// notifyTarget decides where a newly opened incident's notification goes.
|
||||
//
|
||||
// The on-call user's own topic when they have one, otherwise the fallback
|
||||
// topic with no user attached. Deliberately not "the fallback topic, attributed
|
||||
// to the on-call user": the fallback is shared, and an Acknowledge button on a
|
||||
// shared topic would let any subscriber acknowledge as somebody else.
|
||||
func notifyTarget(ctx context.Context, q querier, cfg NotifyConfig, onCall *int64) (topic string, userID *int64) {
|
||||
// The on-call user's own channels when they have any, otherwise the fallback
|
||||
// target with no user attached. Deliberately not "the fallback, attributed to
|
||||
// the on-call user": the fallback is shared, and an Acknowledge button on a
|
||||
// shared target would let any recipient acknowledge as somebody else.
|
||||
func notifyTarget(ctx context.Context, q querier, cfg NotifyConfig, onCall *int64) []recipient {
|
||||
if onCall != nil {
|
||||
var t *string
|
||||
err := q.QueryRowContext(ctx,
|
||||
"SELECT ntfy_topic FROM users WHERE id = $1", *onCall).Scan(&t)
|
||||
if err == nil && t != nil && *t != "" {
|
||||
return *t, onCall
|
||||
if rs, err := userRecipients(ctx, q, cfg, *onCall); err == nil && len(rs) > 0 {
|
||||
return rs
|
||||
}
|
||||
}
|
||||
return cfg.FallbackTopic, nil
|
||||
return fallbackRecipients(cfg, cfg.Fallback)
|
||||
}
|
||||
|
||||
// fallbackRecipients turns a fallback target into the zero or one recipients it
|
||||
// stands for. A target whose provider this server does not have is nobody.
|
||||
func fallbackRecipients(cfg NotifyConfig, t notify.Target) []recipient {
|
||||
if t.Address == "" || !cfg.Providers.Has(t.Type) {
|
||||
return nil
|
||||
}
|
||||
return []recipient{{typ: t.Type, address: t.Address}}
|
||||
}
|
||||
|
||||
// enqueueOpened queues the notification for a freshly opened incident. It is the
|
||||
// only enqueue point that has to resolve a topic from scratch; every later
|
||||
// only enqueue point that has to resolve recipients from scratch; every later
|
||||
// notification for the incident reuses what this one chose.
|
||||
func enqueueOpened(ctx context.Context, q querier, cfg NotifyConfig, incidentID int64, onCall *int64) error {
|
||||
if !cfg.enabled() {
|
||||
return nil
|
||||
}
|
||||
topic, userID := notifyTarget(ctx, q, cfg, onCall)
|
||||
if topic == "" {
|
||||
// Nobody on call has a topic and there is no fallback: there is nowhere
|
||||
recipients := notifyTarget(ctx, q, cfg, onCall)
|
||||
if len(recipients) == 0 {
|
||||
// Nobody on call has a channel and there is no fallback: there is nowhere
|
||||
// to send this, and queueing it would only accumulate undeliverable rows.
|
||||
return nil
|
||||
}
|
||||
return enqueueNotification(ctx, q, incidentID, userID, topic, notifyTriggered)
|
||||
return enqueueNotifications(ctx, q, incidentID, recipients, notifyTriggered)
|
||||
}
|
||||
|
||||
// enqueueResolved queues the all-clear, reusing the topic the incident's last
|
||||
// notification went to. That needs no configuration to reach this function, and
|
||||
// it gives the right rule for free: you only hear that something resolved if you
|
||||
// were told it started.
|
||||
// enqueueResolved queues the all-clear, reusing the recipients of the incident's
|
||||
// last batch. That needs no configuration to reach this function, and it gives
|
||||
// the right rule for free: you only hear that something resolved if you were
|
||||
// told it started. A channel since disabled or removed is not told.
|
||||
func enqueueResolved(ctx context.Context, q querier, incidentID int64) error {
|
||||
var topic string
|
||||
var userID *int64
|
||||
err := q.QueryRowContext(ctx, `
|
||||
SELECT topic, user_id FROM notifications
|
||||
WHERE incident_id = $1 ORDER BY id DESC LIMIT 1`, incidentID).Scan(&topic, &userID)
|
||||
if err == sql.ErrNoRows {
|
||||
return nil
|
||||
}
|
||||
rows, err := q.QueryContext(ctx, `
|
||||
SELECT n.user_id, n.channel_id, n.channel_type, n.address
|
||||
FROM notifications n
|
||||
LEFT JOIN notification_channels c ON c.id = n.channel_id
|
||||
WHERE n.incident_id = $1
|
||||
AND n.created_at = (SELECT MAX(created_at) FROM notifications WHERE incident_id = $1)
|
||||
AND (n.user_id IS NULL OR c.enabled)
|
||||
ORDER BY n.id`, incidentID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return enqueueNotification(ctx, q, incidentID, userID, topic, notifyResolved)
|
||||
|
||||
var recipients []recipient
|
||||
for rows.Next() {
|
||||
var r recipient
|
||||
if err := rows.Scan(&r.userID, &r.channelID, &r.typ, &r.address); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
recipients = append(recipients, r)
|
||||
}
|
||||
err = rows.Err()
|
||||
rows.Close()
|
||||
if err != nil || len(recipients) == 0 {
|
||||
return err
|
||||
}
|
||||
return enqueueNotifications(ctx, q, incidentID, recipients, notifyResolved)
|
||||
}
|
||||
|
||||
// humanDuration renders an age the way a person reads it at 3am: coarse, and
|
||||
|
||||
Reference in New Issue
Block a user