// Package notify tells the web app that a counter reached its limit. // // One URL template per level (org_url, address_url, domain_url) is filled // with the event and fetched with GET + basic auth, in the background, with // retries. The web app (migadu_web GET /api/limits) decides what happens to // the customer; this package only delivers the event. package notify import ( "context" "errors" "fmt" "io" "net/http" "net/url" "strconv" "strings" "sync" "time" ) // Config holds the notification settings. type Config struct { OrgURL string // $LEVEL=O AddressURL string // $LEVEL=A DomainURL string // $LEVEL=D User string // basic auth; empty = no Authorization header Password string Timeout time.Duration // per attempt (default 5s) Attempts int // total attempts (default 5) Backoff time.Duration // first retry delay, doubled each retry (default 1s) MaxConcurrent int // notifications in flight (default 8) } // URLFor returns the template for a level, "" when that level is disabled. func (c Config) URLFor(level string) string { switch level { case "O": return c.OrgURL case "A": return c.AddressURL case "D": return c.DomainURL } return "" } // URLName returns the config key that serves a level, for log lines. func URLName(level string) string { switch level { case "O": return "org_url" case "A": return "address_url" case "D": return "domain_url" } return "" } // Event is one "limit reached" occurrence. type Event struct { TraceID string Route string // in / out, for log correlation Level string // A, D or O ID string // address, domain or organisation id Flow string // i or o Address string // the address that triggered it ($ADDRESS / by=) Count int64 // counter value after the increment ($COUNT) Limit int64 // configured limit ($LIMIT) } // BuildURL fills the template placeholders with URL-escaped event values. // $PERIOD is always "d" (per day). func BuildURL(tmpl string, ev Event) string { esc := url.QueryEscape return strings.NewReplacer( "$LEVEL", esc(ev.Level), "$LIMIT", strconv.FormatInt(ev.Limit, 10), "$ID", esc(ev.ID), "$FLOW", esc(ev.Flow), "$PERIOD", "d", "$ADDRESS", esc(ev.Address), "$COUNT", strconv.FormatInt(ev.Count, 10), ).Replace(tmpl) } // Hooks receive the outcome of every dispatched notification (transaction // log, metrics). Nil hooks are ignored. type Hooks struct { Sent func(ev Event, url string, attempt int) Failed func(ev Event, url string, attempts int, err error) } // Notifier sends events in the background. type Notifier struct { cfg Config hooks Hooks client *http.Client sem chan struct{} wg sync.WaitGroup quit chan struct{} mu sync.Mutex closed bool } // New applies defaults and returns a ready Notifier. func New(cfg Config, hooks Hooks) *Notifier { if cfg.Timeout <= 0 { cfg.Timeout = 5 * time.Second } if cfg.Attempts <= 0 { cfg.Attempts = 5 } if cfg.Backoff <= 0 { cfg.Backoff = time.Second } if cfg.MaxConcurrent <= 0 { cfg.MaxConcurrent = 8 } return &Notifier{ cfg: cfg, hooks: hooks, client: &http.Client{Timeout: cfg.Timeout}, sem: make(chan struct{}, cfg.MaxConcurrent), quit: make(chan struct{}), } } // Notify dispatches ev in the background. It returns false, without doing // anything, when no URL is configured for the level or after Shutdown. func (n *Notifier) Notify(ev Event) bool { tmpl := n.cfg.URLFor(ev.Level) if tmpl == "" { return false } n.mu.Lock() if n.closed { n.mu.Unlock() return false } n.wg.Add(1) n.mu.Unlock() u := BuildURL(tmpl, ev) go func() { defer n.wg.Done() n.sem <- struct{}{} defer func() { <-n.sem }() n.send(ev, u) }() return true } func (n *Notifier) send(ev Event, u string) { var lastErr error for attempt := 1; attempt <= n.cfg.Attempts; attempt++ { err, retry := n.attempt(u) if err == nil { if n.hooks.Sent != nil { n.hooks.Sent(ev, u, attempt) } return } lastErr = err if !retry { if n.hooks.Failed != nil { n.hooks.Failed(ev, u, attempt, err) } return } if attempt < n.cfg.Attempts { select { case <-time.After(n.cfg.Backoff << (attempt - 1)): case <-n.quit: lastErr = fmt.Errorf("shutting down after attempt %d: %w", attempt, err) attempt = n.cfg.Attempts // stop retrying } } } if n.hooks.Failed != nil { n.hooks.Failed(ev, u, n.cfg.Attempts, lastErr) } } // attempt performs one GET. retry is false for 4xx: wrong credentials or a // wrong URL will not fix themselves. func (n *Notifier) attempt(u string) (err error, retry bool) { req, err := http.NewRequest(http.MethodGet, u, nil) if err != nil { return err, false } if n.cfg.User != "" { req.SetBasicAuth(n.cfg.User, n.cfg.Password) } req.Header.Set("User-Agent", "migadu_counters") resp, err := n.client.Do(req) if err != nil { return err, true } defer resp.Body.Close() _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096)) switch { case resp.StatusCode >= 200 && resp.StatusCode < 300: return nil, false case resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500: return fmt.Errorf("unexpected status %d", resp.StatusCode), true default: return fmt.Errorf("unexpected status %d", resp.StatusCode), false } } // Shutdown refuses new events, stops pending retries and waits for in-flight // requests until ctx expires. Safe to call more than once. func (n *Notifier) Shutdown(ctx context.Context) error { n.mu.Lock() if !n.closed { n.closed = true close(n.quit) } n.mu.Unlock() done := make(chan struct{}) go func() { n.wg.Wait() close(done) }() select { case <-done: return nil case <-ctx.Done(): return errors.Join(ctx.Err(), errors.New("notifications still in flight")) } }