Files
memby/server/internal/notify/notify.go
T

344 lines
13 KiB
Go
Raw Normal View History

2026-08-19 06:57:59 +12:00
// Package notify is the one door every outbound notification leaves Memby through.
//
// Before it, each feature both decided to notify somebody and performed the delivery
// itself: the Sonarr lifecycle scanner wrote a row into user_notifications, the library
// ingester pushed a banner into Redis, the integrations dispatcher posted to Discord. Each
// knew how to deliver and none knew that the others existed, so the only way to answer
// "what did Memby send, to whom, and did it work" was to read three subsystems' log lines
// and hope every one of them had logged.
//
// The flow is now
//
// feature/event → notify.Service → Deliverer → notification log
//
// and the audit trail is a property of the door rather than something each feature
// remembers to do. A feature says *what* it wants said and to whom; which provider carries
// it, and the record of what happened, belong here.
//
// Two rules hold the package up:
//
// - **Logging never blocks delivery.** The record is written after the provider has
// already answered, on a context detached from the caller's, and a write that fails is
// logged and swallowed. A notification history that could suppress a notification would
// be worse than no history.
// - **Nothing secret is ever recorded.** A webhook's address is its credential, and an
// Emby token is a live upstream session; neither has any business in a table the
// console renders. Notification carries a Target — a destination's *name* — never its
// address, and Redact is the belt-and-braces pass before anything is stored.
package notify
import (
"context"
"encoding/json"
"log/slog"
"strings"
"time"
)
// Channel is how a notification reaches somebody. It is stored, so it is part of the
// console's vocabulary: adding one means adding a Deliverer and nothing else.
type Channel string
const (
// ChannelInApp is a stored notification in one viewer's own list — My Alerts on the
// television. It follows the person to whichever set they sign into.
ChannelInApp Channel = "in-app"
// ChannelBroadcast is a service alert: the bar every signed-in television draws off
// the /v1/status poll. It has no recipient, because the recipient is the household.
ChannelBroadcast Channel = "broadcast"
// ChannelWebhook is an outbound HTTP delivery to somebody else's service — Discord
// today, and whatever the integrations package learns to speak next.
ChannelWebhook Channel = "webhook"
)
// Status is what became of one notification.
//
// Sent and Delivered are deliberately different answers. Most of Memby's channels are
// stores rather than transports — writing a row into somebody's list is done the moment it
// returns, and there is nobody to acknowledge it — so those report Sent. Delivered is
// reserved for a provider that actually confirmed receipt, which today means a webhook
// that answered 2xx. Collapsing the two would make the console claim an acknowledgement
// that nothing ever gave.
type Status string
const (
StatusSent Status = "sent"
StatusDelivered Status = "delivered"
StatusFailed Status = "failed"
StatusPending Status = "pending"
// StatusSkipped is a notification that was deliberately not delivered, and it is the
// most useful row on the page: a viewer's preferences declined it, a duplicate was
// suppressed by its source key, or an operator has the window switched off. Without
// it, "Memby never told me" and "Memby decided not to tell you" are the same silence.
StatusSkipped Status = "skipped"
)
// Notification is what a feature asks for. It describes the news, never the transport.
type Notification struct {
// Channel selects the provider.
Channel Channel
// Kind is the notification type: "show-return", "watch-time-week", "sonarr-import".
// It is the client's vocabulary too, so it is passed through rather than translated.
Kind string
// Source names the service that decided to send this — "sonarr-lifecycle",
// "watch-time-digest", "library-ingest". It answers "why did this arrive", which the
// kind alone often cannot: two features can legitimately produce the same kind.
Source string
// UserID and Username identify the recipient. Both empty is a household broadcast,
// which is a real answer rather than a missing one.
UserID string
Username string
Title string
Body string
// ItemID links the notification to a title, where there is one.
ItemID string
// Target names a destination that is not a person — an integration's name, for a
// webhook. Never its address: see the package comment.
Target string
// SourceKey is the caller's idempotency key where it has one. It is what lets the
// console explain a skipped row as "already sent" rather than as an unexplained gap.
SourceKey string
// EventAt is when the news happened, where that differs from when it was sent — an
// episode's broadcast time, a digest's period end.
EventAt *time.Time
// Metadata is free-form context for the detail view. Keep it small and keep it
// non-secret; Redact drops anything whose key looks like a credential.
Metadata map[string]any
}
// Outcome is what the provider reported.
type Outcome struct {
Status Status
// Detail is the failure or the reason, and it is the whole value of the detail view:
// "429 Too Many Requests", "the viewer has summaries switched off", "already sent".
Detail string
// Err is the delivery error where there was one, returned to the caller so a feature
// that wants to react to a failure still can. It is never itself the audit trail.
Err error
}
// Sent is the ordinary success for a store-shaped channel.
func Sent() Outcome { return Outcome{Status: StatusSent} }
// Delivered is for a provider that confirmed receipt.
func Delivered(detail string) Outcome { return Outcome{Status: StatusDelivered, Detail: detail} }
// Failed records a delivery that was attempted and did not work.
func Failed(err error) Outcome {
if err == nil {
return Outcome{Status: StatusFailed, Detail: "delivery failed"}
}
return Outcome{Status: StatusFailed, Detail: err.Error(), Err: err}
}
// Skipped records a notification deliberately not delivered, with the reason.
func Skipped(reason string) Outcome { return Outcome{Status: StatusSkipped, Detail: reason} }
// Deliverer is one channel's provider. A new channel is a type implementing this and a
// Register call; nothing else in the package has a case per channel.
type Deliverer interface {
Channel() Channel
Deliver(ctx context.Context, n Notification) Outcome
}
// DelivererFunc adapts a plain function, which is what every provider in the gateway is:
// a small closure over an existing subsystem.
type DelivererFunc struct {
Name Channel
Fn func(ctx context.Context, n Notification) Outcome
}
func (d DelivererFunc) Channel() Channel { return d.Name }
func (d DelivererFunc) Deliver(ctx context.Context, n Notification) Outcome {
return d.Fn(ctx, n)
}
// Record is one row of the audit trail — the notification as it was asked for, plus what
// happened to it.
type Record struct {
OccurredAt time.Time
Channel Channel
Kind string
Source string
UserID string
Username string
Title string
Body string
ItemID string
Target string
SourceKey string
Status Status
Detail string
DurationMS int64
EventAt *time.Time
Metadata json.RawMessage
}
// Recorder is the audit trail's storage. An interface rather than *store.Store so the
// package can be tested without a database, and so a Service built with no recorder — every
// unit test of a feature that notifies — still delivers.
type Recorder interface {
RecordNotification(ctx context.Context, record Record) error
}
// recordTimeout bounds the audit write. It is short on purpose: the notification has
// already been delivered by the time this runs, so a slow database must cost the history
// rather than hold up the feature that produced the news.
const recordTimeout = 5 * time.Second
// Service is the door. One instance, created at start-up.
type Service struct {
recorder Recorder
log *slog.Logger
deliverers map[Channel]Deliverer
now func() time.Time
}
func New(recorder Recorder, log *slog.Logger) *Service {
if log == nil {
log = slog.Default()
}
return &Service{
recorder: recorder,
log: log.With("component", "notify"),
deliverers: map[Channel]Deliverer{},
now: time.Now,
}
}
// Register installs a provider. Called at start-up only; the map is not guarded because
// nothing registers after the first request is served.
func (s *Service) Register(deliverers ...Deliverer) {
if s == nil {
return
}
for _, deliverer := range deliverers {
if deliverer != nil {
s.deliverers[deliverer.Channel()] = deliverer
}
}
}
// Send delivers a notification and records what happened.
//
// The order is the design: deliver, then record. A history written first would be a claim
// rather than a record, and one written inside the delivery path would be able to fail the
// delivery. A Service that is nil, or has no provider for the channel, still answers — a
// feature must never have to nil-check the notification layer.
func (s *Service) Send(ctx context.Context, n Notification) Outcome {
if s == nil {
return Skipped("notifications are not configured")
}
n = Redact(n)
started := s.now()
deliverer, ok := s.deliverers[n.Channel]
var outcome Outcome
if !ok {
// A missing provider is a configuration fault, not a delivery failure, and it is
// worth a row: a console showing every "show-return" as failed on a gateway with
// no in-app provider is what would send somebody looking in the right place.
outcome = Failed(errNoDeliverer{channel: n.Channel})
} else {
outcome = deliverer.Deliver(ctx, n)
}
s.record(ctx, n, outcome, s.now().Sub(started))
return outcome
}
// Log records a notification that some other code path delivered.
//
// It exists for the one producer that cannot reasonably be inverted: the integrations
// dispatcher is a subscriber on the admin event bus with its own queue, pacing and
// transport registry, and routing its deliveries back out through Send would make the
// audit trail the thing that decides what Discord receives. It posts, then says what
// happened. Prefer Send everywhere a feature is the one deciding to notify.
func (s *Service) Log(ctx context.Context, n Notification, outcome Outcome, took time.Duration) {
if s == nil {
return
}
s.record(ctx, Redact(n), outcome, took)
}
func (s *Service) record(ctx context.Context, n Notification, outcome Outcome, took time.Duration) {
if s.recorder == nil {
return
}
record := Record{
OccurredAt: s.now().UTC(),
Channel: n.Channel,
Kind: n.Kind,
Source: n.Source,
UserID: n.UserID,
Username: n.Username,
Title: n.Title,
Body: n.Body,
ItemID: n.ItemID,
Target: n.Target,
SourceKey: n.SourceKey,
Status: outcome.Status,
Detail: outcome.Detail,
DurationMS: took.Milliseconds(),
EventAt: n.EventAt,
}
if len(n.Metadata) > 0 {
if raw, err := json.Marshal(n.Metadata); err == nil {
record.Metadata = raw
}
}
// Detached from the caller's context, for the reason the search recorder is: a
// television that navigated away, or a request that timed out, still sent this.
writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), recordTimeout)
defer cancel()
if err := s.recorder.RecordNotification(writeCtx, record); err != nil {
s.log.Warn("notification not recorded",
"channel", n.Channel, "kind", n.Kind, "status", outcome.Status, "error", err)
}
}
type errNoDeliverer struct{ channel Channel }
func (e errNoDeliverer) Error() string {
return "no delivery provider for the " + string(e.channel) + " channel"
}
// secretish matches a metadata key whose value must never be stored.
//
// The rule is a substring match on the key rather than an inspection of the value,
// deliberately: a token is not recognisable by looking at it, and the one thing a caller
// reliably gets right is what they called the field.
var secretish = []string{
"token", "secret", "password", "apikey", "api_key", "credential",
"webhook", "url", "authorization",
}
// Redact is the last thing between a notification and the audit trail.
//
// Callers are already expected not to put a credential in a Notification — Target is a
// destination's name and never its address — and this is what makes that a property of the
// package rather than of every caller's diligence.
func Redact(n Notification) Notification {
if len(n.Metadata) == 0 {
return n
}
cleaned := make(map[string]any, len(n.Metadata))
for key, value := range n.Metadata {
if isSecretKey(key) {
continue
}
cleaned[key] = value
}
n.Metadata = cleaned
return n
}
func isSecretKey(key string) bool {
lowered := strings.ToLower(key)
for _, needle := range secretish {
if strings.Contains(lowered, needle) {
return true
}
}
return false
}