ARTEX/server/notifier.go

package server

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/Autumn-27/artex/db"
	"github.com/Autumn-27/artex/notify"
)

// Global setting keys (stored in the settings key-value table, no dedicated table needed).
const (
	// settingNotifyEnabled is the master switch for delivery. Default on: it exists
	// as a one-shot kill switch during maintenance, not as the enable condition for
	// the feature — the real enable condition is "whether any channel is configured".
	settingNotifyEnabled = "notify_enabled"
	// settingNotifyPublicBaseURL is the externally reachable address used to build
	// detail back-links for vulnerabilities (e.g. https://artex.example.com). If left
	// empty, messages carry no back-link button. There is no reusable external-address
	// setting in the project, so a new one is added here.
	settingNotifyPublicBaseURL = "notify_public_base_url"
	// settingNotifyDigestMinutes is the digest-mode period (minutes).
	settingNotifyDigestMinutes = "notify_digest_interval_min"
)

const (
	// notifyTick is the delivery engine's polling interval. 3 seconds is the upper
	// bound of this engine's realtimeness, and also the main source of latency
	// between "vulnerability persisted" and "message arrives at the IM".
	notifyTick = 3 * time.Second
	// notifyLease is the lease duration when claiming deliveries. It must be
	// significantly longer than the worst-case duration of a single delivery
	// (the notify package's HTTP client timeout is 15 seconds), otherwise the
	// same row could be delivered by two dispatchers at the same time.
	notifyLease = 3 * time.Minute
	// notifyFanOutPerTick limits the number of events fanned out per tick, so that
	// enabling a channel for the first time does not expand the entire historical
	// backlog into delivery tasks in one go.
	notifyFanOutPerTick = 200
	// notifyDefaultDigestMinutes is the default digest period.
	notifyDefaultDigestMinutes = 30
	// notifyUnlimitedBurstPerTick is the per-tick delivery cap when a channel has
	// no rate limit configured. It exists to prevent "one channel configured as
	// unlimited + a single scan finding thousands of vulnerabilities" from dragging
	// a single tick into a long blocking loop.
	notifyUnlimitedBurstPerTick = 50
	// notifyMaxSendsPerChannelPerTick is the maximum number of deliveries per tick
	// for a single channel.
	//
	// This cap is derived backwards from the **lease duration**: claiming marks rows
	// with a lease (notifyLease = 3 minutes); if one tick serially delivers so many
	// rows that the worst-case duration exceeds the lease, the lease on the later
	// rows expires before they are even sent. Within a single process this is
	// harmless (Run is a single serial goroutine, ticks never re-enter), but with
	// **two processes on the same database**, the peer will reclaim rows whose
	// leases expired and resend them, double-increment attempts, and mark them
	// failed while the original process is still delivering.
	//
	// Value: 3-minute lease / 30-second per-send timeout = 6, which **consumes the
	// lease exactly** with zero headroom, so it cannot be used; 5 leaves a 30-second
	// margin on a 150-second worst case. This relationship is pinned down by
	// TestNotifyTickBudgetFitsWithinLease — changing any of notifyLease,
	// notifySendTimeout, or this value will fail that assertion.
	notifyMaxSendsPerChannelPerTick = 5
	// notifySendTimeout is the timeout for a single delivery. It also determines
	// the value of the constant above; the two multiplied together must not exceed
	// notifyLease, see TestNotifyTickBudgetFitsWithinLease.
	notifySendTimeout = 30 * time.Second
)

// notifyBackoff is the backoff sequence for failed-delivery retries, indexed by
// the number of attempts already made. 3 chances (including the first attempt)
// correspond to db.MaxNotifyAttempts; the two must be changed together.
var notifyBackoff = []time.Duration{
	time.Second,
	5 * time.Second,
	30 * time.Second,
}

// Notifier is the delivery engine for vulnerability notifications.
//
// It runs as an independent goroutine alongside the Scheduler (see server.New).
// It deliberately does not reuse the Scheduler's tick: delivery's realtimeness
// requirement (3 seconds) differs from the trigger's business cadence, and their
// failures must not affect each other — a stuck delivery must not impact agent
// triggering.
type Notifier struct {
	s  *Server
	pg *db.DB

	// mu guards buckets. The number of channels is small and contention is low,
	// so a single mutex is enough; a finer-grained structure is not worth it.
	mu      sync.Mutex
	buckets map[int64]*notifyBucket
}

// notifyBucket is the token bucket for a single channel.
//
// A token bucket is used instead of a sliding window of "count then reset each
// minute", because the latter has terrible boundary effects: sending the full 20
// messages at the end of a window and another 20 right after means 40 messages in
// one second from the platform's perspective, which gets rate-limited; a token
// bucket refills at a constant rate and inherently avoids such bursts.
type notifyBucket struct {
	tokens   float64
	lastFill time.Time
}

func newNotifier(s *Server) *Notifier {
	return &Notifier{s: s, pg: s.m.pg, buckets: map[int64]*notifyBucket{}}
}

// Run loops until ctx ends. Started once by server.New.
func (n *Notifier) Run(ctx context.Context) {
	if n.pg == nil {
		return
	}
	t := time.NewTicker(notifyTick)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			n.step(ctx)
		}
	}
}

// step runs one round: first fan out new events, then deliver due tasks.
//
// Failure of any step only logs and never interrupts the loop — a failure in the
// notification system must never escalate into a process-level problem. Every
// tick is independent, and the next round will naturally retry.
func (n *Notifier) step(ctx context.Context) {
	if !n.enabled() {
		return
	}
	if _, _, err := n.pg.FanOutPendingEvents(ctx, notifyFanOutPerTick); err != nil {
		log.Printf("[notify] failed to fan out events: %v", err)
		return
	}
	channels, err := n.pg.ListNotificationChannels(ctx)
	if err != nil {
		log.Printf("[notify] failed to read channels: %v", err)
		return
	}
	baseURL := n.publicBaseURL()
	for _, ch := range channels {
		if !ch.IsEnabled() {
			continue
		}
		// The token bucket is metered in **messages** (equivalent to HTTP requests),
		// not vulnerabilities. In realtime mode the two are the same (one message per
		// vulnerability); in digest mode a whole batch of vulnerabilities becomes one
		// message, so it consumes only one token.
		//
		// Both modes ask the token bucket first and then claim within that allowance —
		// the order must not be reversed, otherwise deliveries blocked by rate limiting
		// would already have consumed retry attempts.
		now := time.Now()
		if ch.Mode == db.NotifyModeDigest {
			tokens, claimLimit := digestTickPlan()
			if n.takeTokens(ch.ID, ch.RatePerMin, tokens, now) <= 0 {
				continue
			}
			n.stepDigest(ctx, ch, claimLimit, baseURL)
			continue
		}
		allow := n.takeTokens(ch.ID, ch.RatePerMin, notifyMaxSendsPerChannelPerTick, now)
		if allow <= 0 {
			continue
		}
		n.stepRealtime(ctx, ch, allow, baseURL)
	}
}

// digestTickPlan returns the token consumption and batch-size upper bound for a
// digest channel in this tick.
//
// The two return values are **two different dimensions**, which is exactly why
// this is a separate function:
//
//   - tokens is the number of messages. A batch of vulnerabilities becomes one
//     message and one HTTP request, so it is always 1. rate_per_min therefore
//     still applies to digests (at most this many digest messages per minute).
//   - claimLimit is how many vulnerabilities a batch may hold at most. It is
//     bounded only by memory and is unrelated to the request budget.
//
// A previous version, in order to make rate_per_min apply to digests, passed the
// per-tick request budget (notifyMaxSendsPerChannelPerTick, derived backwards
// from the lease) directly as the batch size. The consequence was that a channel
// with rate_per_min=20 only refilled 1 token in a 3-second tick, so every digest
// message carried just 1 vulnerability — the digest degenerated into "realtime
// delivery with digest wording", readers received a stream of "1 new vulnerability
// in the last 30 minutes", and db.MaxDigestBatchSize was never reachable.
//
// This symptom is hard to catch in end-to-end tests (existing cases all pass a
// large enough limit to stepDigest manually, bypassing the allowance calculation
// in step), so the decision is pinned down here directly by
// TestDigestTickPlanDecouplesBatchSizeFromSendBudget.
func digestTickPlan() (tokens, claimLimit int) {
	return 1, db.MaxDigestBatchSize
}

// stepRealtime claims and delivers a channel's realtime tasks, one message per
// vulnerability.
func (n *Notifier) stepRealtime(ctx context.Context, ch *db.NotificationChannel, allow int, baseURL string) {
	deliveries, err := n.pg.ClaimRealtimeDeliveries(ctx, ch.ID, allow, notifyLease)
	if err != nil {
		log.Printf("[notify] failed to claim realtime deliveries channel=%d: %v", ch.ID, err)
		return
	}
	if len(deliveries) == 0 {
		return
	}
	channel, cfg, ok := n.adapt(ch)
	if !ok {
		_ = n.pg.FailDeliveries(ctx, deliveryIDs(deliveries), fmt.Sprintf("channel type %q is not registered", ch.Kind))
		return
	}
	for _, dl := range deliveries {
		msg, err := n.renderSingle(ctx, dl, baseURL)
		if err != nil {
			// A render failure is a local data problem; retrying will not help.
			_ = n.pg.FailDeliveries(ctx, []int64{dl.ID}, err.Error())
			continue
		}
		n.send(ctx, channel, cfg, msg, []*db.NotificationDelivery{dl})
	}
}

// stepDigest aggregates a channel's pending deliveries into one message and sends
// it when the batch is due.
func (n *Notifier) stepDigest(ctx context.Context, ch *db.NotificationChannel, allow int, baseURL string) {
	window := n.digestInterval()
	due, err := n.pg.DigestBatchDue(ctx, ch.ID, window)
	if err != nil {
		log.Printf("[notify] failed to check digest batch channel=%d: %v", ch.ID, err)
		return
	}
	if !due {
		return
	}
	deliveries, err := n.pg.ClaimDigestBatch(ctx, ch.ID, allow, notifyLease)
	if err != nil {
		log.Printf("[notify] failed to claim digest batch channel=%d: %v", ch.ID, err)
		return
	}
	if len(deliveries) == 0 {
		return
	}
	channel, cfg, ok := n.adapt(ch)
	if !ok {
		_ = n.pg.FailDeliveries(ctx, deliveryIDs(deliveries), fmt.Sprintf("channel type %q is not registered", ch.Kind))
		return
	}
	msg, included, err := n.renderBatch(ctx, deliveries, baseURL, int(window.Minutes()))
	if err != nil {
		_ = n.pg.FailDeliveries(ctx, deliveryIDs(deliveries), err.Error())
		return
	}
	// Deliveries whose snapshots were broken and did not make it into the message
	// must be marked failed explicitly. Otherwise they stay outside `included`,
	// entering neither the message nor the failure list — when the send succeeds,
	// the later batch marking would miss them, and they would stay in `sending`
	// forever until their leases expire and they get reclaimed over and over.
	if skipped := excludeDeliveries(deliveries, included); len(skipped) > 0 {
		reason := "event snapshot could not be parsed; this vulnerability cannot be rendered into a message"
		if fErr := n.pg.FailDeliveries(ctx, deliveryIDs(skipped), reason); fErr != nil {
			log.Printf("[notify] failed to mark broken-snapshot deliveries failed channel=%s ids=%v: %v", ch.Kind, deliveryIDs(skipped), fErr)
		}
		log.Printf("[notify] skipped %d deliveries with unparseable snapshots channel=%d", len(skipped), ch.ID)
	}
	// Only pass those that made it into the message to send: included[i]
	// corresponds strictly to msg.Items[i], and send relies on this correspondence
	// to attribute "the channel reported fitting the first K entries" to the
	// correct delivery rows.
	n.send(ctx, channel, cfg, msg, included)
}

// send delivers and transitions state according to the result.
//
// A batch of deliveries (possibly dozens in digest mode) shares a single send
// result: either delivered, or retried as a whole batch. There is no per-item
// retry — a digest message is one message, and resending part of it would corrupt
// the batch semantics.
//
// The only exception is **segmentation caused by the channel's length limit**: if
// the channel reports that only the first K entries actually fit, then from entry
// K+1 onward must be left for the next batch instead of being marked succeeded
// along with the rest. Otherwise the truncated vulnerabilities would be neither
// in the message nor in the failure list — they would disappear entirely.
func (n *Notifier) send(ctx context.Context, channel notify.Channel, cfg map[string]any, msg notify.Message, deliveries []*db.NotificationDelivery) {
	// Cap the duration of a single delivery so one stuck channel cannot hold up
	// all remaining channels in this round.
	sendCtx, cancel := context.WithTimeout(ctx, notifySendTimeout)
	defer cancel()
	delivered, err := channel.Send(sendCtx, cfg, msg)
	if err == nil && delivered > 0 {
		if delivered > len(deliveries) {
			// The channel cannot possibly report more delivered entries than there
			// are deliveries; if it does, the render layer miscounted. Treat it as
			// fully delivered and record the problem — better than corrupting the
			// records.
			log.Printf("[notify] channel reported %d delivered entries, more than the %d deliveries channel=%s; treating as fully delivered",
				delivered, len(deliveries), channel.Kind())
			delivered = len(deliveries)
		}
		sent, rest := deliveries[:delivered], deliveries[delivered:]
		if err := n.pg.MarkDeliveriesSent(ctx, deliveryIDs(sent)); err != nil {
			log.Printf("[notify] failed to mark as sent channel=%s ids=%v: %v", channel.Kind(), deliveryIDs(sent), err)
		}
		if len(rest) > 0 {
			// This message has reached the channel's length limit: the remainder
			// re-queues immediately and the next tick continues sending. Use
			// DeferDeliveries rather than RescheduleDeliveries — this is not a
			// failure and must not consume the retry budget (claiming already
			// optimistically incremented attempts by +1, which this will undo).
			if err := n.pg.DeferDeliveries(ctx, deliveryIDs(rest),
				fmt.Sprintf("this message reached the channel length limit; only the first %d entries were delivered, the rest are left for the next batch", delivered)); err != nil {
				log.Printf("[notify] failed to queue segmented continuation channel=%s ids=%v: %v", channel.Kind(), deliveryIDs(rest), err)
			}
		}
		return
	}
	if err == nil {
		// The channel neither reported an error nor said how many entries were
		// delivered. Treat it as a failure (with backoff), so this delivery is not
		// repeatedly claimed yet never marked.
		err = fmt.Errorf("channel did not report a delivered count (delivered=%d)", delivered)
	}

	// Failure handling is decided **per delivery**, not by the maximum attempt
	// count of the whole batch.
	//
	// A previous version failed the entire batch with
	// `if maxAttempts(deliveries) >= MaxNotifyAttempts`, but the attempt counts of
	// items in a batch differ: an old delivery already retried twice (attempts=2)
	// would drag a brand-new delivery (attempts=1) in the same batch into failed —
	// a new vulnerability permanently lost without using a single retry, the exact
	// opposite of the intent of "don't let old rows sink new rows".
	permanent := notify.IsPermanent(err)
	var failIDs, exhaustedIDs []int64
	byDelay := map[time.Duration][]int64{}
	for _, dl := range deliveries {
		switch {
		case permanent:
			failIDs = append(failIDs, dl.ID)
		case dl.Attempts >= db.MaxNotifyAttempts:
			exhaustedIDs = append(exhaustedIDs, dl.ID)
		default:
			delay := notifyBackoff[min(dl.Attempts, len(notifyBackoff)-1)]
			byDelay[delay] = append(byDelay[delay], dl.ID)
		}
	}

	if len(failIDs) > 0 {
		if fErr := n.pg.FailDeliveries(ctx, failIDs, err.Error()); fErr != nil {
			log.Printf("[notify] error marking failed state channel=%s ids=%v: %v", channel.Kind(), failIDs, fErr)
		}
	}
	if len(exhaustedIDs) > 0 {
		reason := fmt.Sprintf("still failing after %d retries: %s", db.MaxNotifyAttempts, err)
		if fErr := n.pg.FailDeliveries(ctx, exhaustedIDs, reason); fErr != nil {
			log.Printf("[notify] error marking failed state channel=%s ids=%v: %v", channel.Kind(), exhaustedIDs, fErr)
		}
	}
	// Reschedule grouped by delay: with only 3 backoff tiers the number of groups
	// is inherently small, so there is no need to issue a separate UPDATE per row
	// (which would turn a 500-row batch into 500 round trips).
	for delay, group := range byDelay {
		if rErr := n.pg.RescheduleDeliveries(ctx, group, delay, err.Error()); rErr != nil {
			log.Printf("[notify] failed to reschedule deliveries channel=%s ids=%v: %v", channel.Kind(), group, rErr)
		}
	}
	if len(failIDs)+len(exhaustedIDs) > 0 {
		log.Printf("[notify] delivery failed channel=%d kind=%s permanent=%d retries-exhausted=%d pending-retry=%d: %s",
			deliveries[0].ChannelID, channel.Kind(), len(failIDs), len(exhaustedIDs), len(byDelay), err)
	}
}

// excludeDeliveries returns the items in all that are not in keep (compared by
// pointer identity). Used to find deliveries that "did not make it into the
// message" — they must be handled explicitly and must not be left in a gray zone.
func excludeDeliveries(all, keep []*db.NotificationDelivery) []*db.NotificationDelivery {
	inKeep := make(map[*db.NotificationDelivery]bool, len(keep))
	for _, dl := range keep {
		inKeep[dl] = true
	}
	var out []*db.NotificationDelivery
	for _, dl := range all {
		if !inKeep[dl] {
			out = append(out, dl)
		}
	}
	return out
}

// adapt resolves the channel implementation and parses its configuration.
// Returning ok=false means the type is not registered; deliveries should be
// marked failed directly instead of retried indefinitely.
func (n *Notifier) adapt(ch *db.NotificationChannel) (notify.Channel, map[string]any, bool) {
	channel, ok := notify.Get(ch.Kind)
	if !ok {
		return nil, nil, false
	}
	var cfg map[string]any
	if len(ch.Config) > 0 {
		// When config parsing fails, fall back to an empty map: the channel's own
		// Validate will report "which field is missing", an error far more
		// actionable for the user than a JSON parse error.
		_ = json.Unmarshal(ch.Config, &cfg)
	}
	if cfg == nil {
		cfg = map[string]any{}
	}
	return channel, cfg, true
}

// renderSingle renders a single-vulnerability message.
func (n *Notifier) renderSingle(ctx context.Context, dl *db.NotificationDelivery, baseURL string) (notify.Message, error) {
	snap, err := parseSnapshot(dl)
	if err != nil {
		return notify.Message{}, err
	}
	item, err := n.itemFor(ctx, snap, baseURL)
	if err != nil {
		return notify.Message{}, err
	}
	return notify.Message{Items: []notify.Item{item}, HomeURL: baseURL}, nil
}

// renderBatch renders a digest message. Snapshots are parsed one by one — a
// single broken one only skips that entry and must not sink the whole digest.
//
// The return value included corresponds **strictly one-to-one** with msg.Items
// (the i-th delivery ↔ the i-th item). This correspondence is a hard requirement:
// the caller marks the first K deliveries as sent based on "the channel reported
// fitting the first K entries". If a broken snapshot were skipped here without
// removing the skipped delivery from included, the indices would shift — a broken
// entry that should have failed would be marked sent, and a good entry would be
// misjudged as undelivered. The broken ones are marked failed explicitly by the
// caller, see stepDigest.
func (n *Notifier) renderBatch(ctx context.Context, deliveries []*db.NotificationDelivery, baseURL string, windowMinutes int) (notify.Message, []*db.NotificationDelivery, error) {
	items := make([]notify.Item, 0, len(deliveries))
	included := make([]*db.NotificationDelivery, 0, len(deliveries))
	for _, dl := range deliveries {
		snap, err := parseSnapshot(dl)
		if err != nil {
			// A broken snapshot goes neither into the message nor into included —
			// handling it is the caller's responsibility (explicitly marking it
			// failed, rather than sneaking it through among the "sent" ones).
			log.Printf("[notify] skipping unparseable snapshot in digest batch delivery=%d: %v", dl.ID, err)
			continue
		}
		item, err := n.itemFor(ctx, snap, baseURL)
		if err != nil {
			return notify.Message{}, nil, err
		}
		items = append(items, item)
		included = append(included, dl)
	}
	if len(items) == 0 {
		return notify.Message{}, nil, fmt.Errorf("all %d deliveries in the digest batch failed to parse", len(deliveries))
	}
	return notify.Message{
		Items:         items,
		Batch:         true,
		WindowMinutes: windowMinutes,
		HomeURL:       baseURL,
	}, included, nil
}

// itemFor renders an event snapshot into a notification item, also resolving
// asset names and the detail back-link.
func (n *Notifier) itemFor(ctx context.Context, snap notify.Snapshot, baseURL string) (notify.Item, error) {
	assets, err := n.pg.NotificationAssetNames(ctx, snap.AssetIDs)
	if err != nil {
		// Failing to resolve asset names must not block delivery: missing names is
		// far less severe than missing the notification — the message just has one
		// fewer asset line.
		log.Printf("[notify] failed to resolve asset names finding=%d: %v", snap.FindingID, err)
	}
	item := notify.Item{
		FindingID:  snap.FindingID,
		Name:       snap.Name,
		VulnClass:  snap.VulnClass,
		Severity:   snap.Severity,
		Summary:    snap.Summary,
		Assets:     assets,
		FromStatus: snap.FromStatus,
		ToStatus:   snap.ToStatus,
	}
	if baseURL != "" {
		// The detail page route is in web/src/app/(main)/function/findings/detail/page.tsx;
		// it reads the finding id from the `id` query parameter.
		item.DetailURL = fmt.Sprintf("%s/function/findings/detail?id=%d", baseURL, snap.FindingID)
	}
	return item, nil
}

// takeTokens takes **at most want** tokens from the channel's token bucket and
// returns the number actually taken.
//
// One token = one message (one HTTP request). In realtime mode the caller passes
// as many as it needs; in digest mode a whole batch of vulnerabilities sends only
// one message, so pass 1.
//
// The bucket capacity is the channel's per-minute limit and refills at a constant
// rate. ratePerMin<=0 means no rate limit, in which case a finite but
// sufficiently large value is returned, to prevent a single tick from being
// dragged down by unlimited backlog.
//
// The `want` cap is necessary: without it the only option would be draining the
// whole bucket, while the caller has its own per-tick cap — the extra tokens are
// both unusable and gone before the next refill. Accumulated burst capacity would
// be forever unreachable, and even "no pending deliveries this round" would still
// be charged.
func (n *Notifier) takeTokens(channelID int64, ratePerMin, want int, now time.Time) int {
	if want <= 0 {
		return 0
	}
	if ratePerMin <= 0 {
		return min(want, notifyUnlimitedBurstPerTick)
	}
	n.mu.Lock()
	defer n.mu.Unlock()
	b := n.buckets[channelID]
	if b == nil {
		b = &notifyBucket{tokens: float64(ratePerMin), lastFill: now}
		n.buckets[channelID] = b
	}
	// Refill by real elapsed time at a rate of ratePerMin/60 per second.
	if elapsed := now.Sub(b.lastFill).Seconds(); elapsed > 0 {
		b.tokens = minF(float64(ratePerMin), b.tokens+elapsed*float64(ratePerMin)/60)
		b.lastFill = now
	}
	// Add a tiny epsilon before truncating: the token count is accumulated in
	// floating point, and two half refills can yield 0.5 + 0.5 = 0.9999999999,
	// which a plain int() would truncate to 0 — a mathematically full bucket that
	// yields no token. 1e-9 is far smaller than one token and will not let a real
	// shortfall through.
	take := min(int(b.tokens+1e-9), want)
	if take <= 0 {
		return 0
	}
	b.tokens -= float64(take)
	return take
}

// enabled reads the master switch.
func (n *Notifier) enabled() bool {
	return n.pg.GetBool(settingNotifyEnabled, true)
}

// publicBaseURL returns the external address for back-links, with trailing
// slashes removed.
func (n *Notifier) publicBaseURL() string {
	v, ok, err := n.pg.GetSetting(settingNotifyPublicBaseURL)
	if err != nil || !ok {
		return ""
	}
	return trimTrailingSlash(v)
}

// digestInterval returns the digest period, falling back to the default when the
// value is invalid or not configured.
func (n *Notifier) digestInterval() time.Duration {
	v, ok, err := n.pg.GetSetting(settingNotifyDigestMinutes)
	if err != nil || !ok {
		return time.Duration(notifyDefaultDigestMinutes) * time.Minute
	}
	m := 0
	if _, err := fmt.Sscanf(v, "%d", &m); err != nil || m <= 0 {
		return time.Duration(notifyDefaultDigestMinutes) * time.Minute
	}
	return time.Duration(m) * time.Minute
}

// parseSnapshot parses the snapshot of the event a delivery corresponds to.
func parseSnapshot(dl *db.NotificationDelivery) (notify.Snapshot, error) {
	var snap notify.Snapshot
	if len(dl.Snapshot) == 0 {
		return snap, fmt.Errorf("event snapshot of delivery %d is empty", dl.ID)
	}
	if err := json.Unmarshal(dl.Snapshot, &snap); err != nil {
		return snap, fmt.Errorf("failed to parse event snapshot of delivery %d: %w", dl.ID, err)
	}
	if snap.Kind == "" {
		// The event kind is authoritative on the event row; the copy in the
		// snapshot may have been written by an older version.
		snap.Kind = dl.EventKind
	}
	return snap, nil
}

func deliveryIDs(deliveries []*db.NotificationDelivery) []int64 {
	out := make([]int64, 0, len(deliveries))
	for _, dl := range deliveries {
		out = append(out, dl.ID)
	}
	return out
}

func trimTrailingSlash(s string) string {
	for len(s) > 0 && s[len(s)-1] == '/' {
		s = s[:len(s)-1]
	}
	return s
}

func minF(a, b float64) float64 {
	if a < b {
		return a
	}
	return b
}