MealQuest/internal/notify/worker.go

package notify

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"time"

	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db"
	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
)

// Delivery timing.
const (
	// retryFirst is the wait after a row's first failed delivery; each
	// further failure doubles it, up to retryMax.
	retryFirst = time.Minute
	retryMax   = time.Hour
	// giveUpAfter is how long after its creation a row may be delivered.
	// The last retry is due at that moment; a row claimed then or later is
	// marked failed without being sent, and a delivery failing at or
	// after it marks the row failed.
	giveUpAfter = 24 * time.Hour
	// sendTimeout bounds one Send.
	sendTimeout = 30 * time.Second
	// leaseFor is how long a claimed row is not due again. It outlasts
	// sendTimeout and the final write, so a row is only claimed again when
	// the process that claimed it stopped before recording the delivery.
	leaseFor = 2 * time.Minute
	// pollRetryMin and pollRetryMax bound the backoff after a database
	// error while claiming rows.
	pollRetryMin = time.Second
	pollRetryMax = time.Minute
)

// Message is one notification for a Sender to deliver.
type Message struct {
	// ID is the notification_outbox row id.
	ID string
	// EventKey is the event's idempotency key.
	EventKey string
	Channel  Channel
	Payload  Payload
	// Attempt is this delivery's number, from 1.
	Attempt int64
}

// ErrUndeliverable is wrapped by a Sender's error when retrying cannot
// help, such as a channel refusing the message outright: the worker marks
// the row failed at once instead of retrying it.
var ErrUndeliverable = errors.New("notify: undeliverable")

// Sender delivers notifications on one channel. Send returns nil once the
// channel has accepted m, and an error otherwise; the worker retries it,
// unless the error wraps ErrUndeliverable.
// The error's text is stored in notification_outbox.last_error and logged,
// so it must never carry a secret: no webhook URL, key or token (a
// *url.Error quotes its URL).
type Sender interface {
	Send(ctx context.Context, m Message) error
}

// Register makes sender the Sender of ch. It must be called before Start,
// once per channel.
func (s *Service) Register(ch Channel, sender Sender) error {
	if ch != ChannelPush && ch != ChannelChat {
		return fmt.Errorf("notify: register: unknown channel %q", ch)
	}
	if sender == nil {
		return fmt.Errorf("notify: register %s: nil sender", ch)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.started {
		return fmt.Errorf("notify: register %s: worker already started", ch)
	}
	if _, ok := s.senders[ch]; ok {
		return fmt.Errorf("notify: register %s: channel already has a sender", ch)
	}
	s.senders[ch] = sender
	return nil
}

// Start starts the worker. It stops when ctx is cancelled; Wait waits for
// it. A row whose delivery is cut short by ctx keeps its lease and is
// delivered again once the lease ends.
func (s *Service) Start(ctx context.Context) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.started {
		return errors.New("notify: worker already started")
	}
	s.started = true
	s.wg.Add(1)
	go s.work(ctx)
	return nil
}

// Wait blocks until the worker has stopped.
func (s *Service) Wait() {
	s.wg.Wait()
}

// Wake tells the worker to look for due rows, as after EmitTx's
// transaction commits.
func (s *Service) Wake() {
	select {
	case s.wake <- struct{}{}:
	default: // the worker already has a token and will look again
	}
}

// work delivers due rows until ctx is cancelled, then sleeps until the next
// pending row is due or Emit stores one. A database error is retried with
// backoff.
func (s *Service) work(ctx context.Context) {
	defer s.wg.Done()
	backoff := pollRetryMin
	for ctx.Err() == nil {
		wait, err := s.deliverDue(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return
			}
			s.logger.ErrorContext(ctx, "deliver notifications failed; retrying", "error", err, "retry_in", backoff)
			wait, backoff = backoff, min(2*backoff, pollRetryMax)
		} else {
			backoff = pollRetryMin
		}
		var timeout <-chan time.Time
		var timer *time.Timer
		if wait >= 0 {
			timer = time.NewTimer(wait)
			timeout = timer.C
		}
		select {
		case <-s.wake:
		case <-timeout:
		case <-ctx.Done():
		}
		if timer != nil {
			timer.Stop()
		}
	}
}

// deliverDue delivers every row due now and returns how long until the next
// pending row is due, or -1 when none is pending.
func (s *Service) deliverDue(ctx context.Context) (time.Duration, error) {
	q := db.New(s.st.Writer)
	for ctx.Err() == nil {
		now := s.now()
		lease := sql.NullString{String: timestamp(now.Add(leaseFor)), Valid: true}
		row, err := q.ClaimDueNotification(ctx, db.ClaimDueNotificationParams{LeaseUntil: lease, Now: timestamp(now)})
		if errors.Is(err, sql.ErrNoRows) {
			break
		}
		if err != nil {
			return 0, fmt.Errorf("notify: claim notification: %w", err)
		}
		s.deliver(ctx, row)
	}
	if err := ctx.Err(); err != nil {
		return 0, err
	}
	next, err := db.New(s.st.Reader).NextNotificationDue(ctx)
	if errors.Is(err, sql.ErrNoRows) {
		return -1, nil
	}
	if err != nil {
		return 0, fmt.Errorf("notify: read next due notification: %w", err)
	}
	due, err := parseTimestamp(next.String)
	if err != nil {
		return 0, fmt.Errorf("notify: next due notification: %w", err)
	}
	return max(due.Sub(s.now()), 0), nil
}

// deliver sends a claimed row through its channel's sender and records the
// outcome. A row that cannot be sent at all, because its payload does not
// decode or its channel has no sender, is failed at once: retrying cannot
// help. A row 24 hours or more past its creation is failed without being
// sent: it is too stale to deliver.
func (s *Service) deliver(ctx context.Context, row db.NotificationOutbox) {
	logger := s.logger.With("notification_id", row.ID, "event_key", row.EventKey, "channel", row.Channel)
	var payload Payload
	if err := json.Unmarshal([]byte(row.Payload), &payload); err != nil {
		logger.ErrorContext(ctx, "notification failed", "reason", "payload does not decode")
		s.giveUp(ctx, logger, row, 0, "payload does not decode")
		return
	}
	s.mu.Lock()
	sender, ok := s.senders[Channel(row.Channel)]
	s.mu.Unlock()
	if !ok {
		reason := fmt.Sprintf("no sender registered for channel %s", row.Channel)
		logger.ErrorContext(ctx, "notification failed", "reason", reason)
		s.giveUp(ctx, logger, row, 0, reason)
		return
	}

	created, err := parseTimestamp(row.CreatedAt)
	if err != nil {
		reason := fmt.Sprintf("created_at does not parse: %v", err)
		logger.ErrorContext(ctx, "notification failed", "reason", reason)
		s.giveUp(ctx, logger, row, 0, reason)
		return
	}
	deadline := created.Add(giveUpAfter)
	if !s.now().Before(deadline) {
		reason := "not delivered within 24 hours of creation"
		if row.LastError.Valid {
			reason += "; last error: " + row.LastError.String
		}
		logger.ErrorContext(ctx, "notification failed; giving up", "reason", reason)
		s.giveUp(ctx, logger, row, 0, reason)
		return
	}

	attempt := row.Attempts + 1
	sendCtx, cancel := context.WithTimeout(ctx, sendTimeout)
	err = sender.Send(sendCtx, Message{
		ID: row.ID, EventKey: row.EventKey, Channel: Channel(row.Channel), Payload: payload, Attempt: attempt,
	})
	cancel()
	now := s.now()
	if err == nil {
		s.record(ctx, logger, "record notification sent", func(write context.Context) (int64, error) {
			return db.New(s.st.Writer).MarkNotificationSent(write, db.MarkNotificationSentParams{
				Now: timestamp(now), ID: row.ID, LeaseUntil: row.NextAttemptAt,
			})
		})
		logger.InfoContext(ctx, "notification sent", "attempt", attempt)
		return
	}
	if ctx.Err() != nil {
		logger.WarnContext(context.WithoutCancel(ctx), "notification delivery interrupted by shutdown; it is retried when its lease ends")
		return
	}

	if errors.Is(err, ErrUndeliverable) || !now.Before(deadline) {
		logger.ErrorContext(ctx, "notification failed; giving up", "attempt", attempt, "error", err)
		s.giveUp(ctx, logger, row, 1, err.Error())
		return
	}
	next := now.Add(retryDelay(attempt))
	if next.After(deadline) {
		next = deadline
	}
	logger.WarnContext(ctx, "notification delivery failed; retrying", "attempt", attempt, "error", err, "retry_at", timestamp(next))
	s.record(ctx, logger, "record notification retry", func(write context.Context) (int64, error) {
		return db.New(s.st.Writer).RetryNotification(write, db.RetryNotificationParams{
			NextAttemptAt: sql.NullString{String: timestamp(next), Valid: true},
			LastError:     sql.NullString{String: err.Error(), Valid: true},
			Now:           timestamp(now),
			ID:            row.ID,
			LeaseUntil:    row.NextAttemptAt,
		})
	})
}

// giveUp marks a claimed row failed with reason. attempted is 1 when a
// delivery was attempted, else 0.
func (s *Service) giveUp(ctx context.Context, logger *slog.Logger, row db.NotificationOutbox, attempted int64, reason string) {
	now := s.now()
	s.record(ctx, logger, "record notification failed", func(write context.Context) (int64, error) {
		return db.New(s.st.Writer).FailNotification(write, db.FailNotificationParams{
			Attempted:  attempted,
			LastError:  sql.NullString{String: reason, Valid: true},
			Now:        timestamp(now),
			ID:         row.ID,
			LeaseUntil: row.NextAttemptAt,
		})
	})
}

// record makes a row's final write for one delivery, even when ctx has been
// cancelled, bounded by jobs.FinalWriteTimeout. A write that fails, or that
// finds the row no longer leased to this claim, is logged; the row keeps
// its lease and is delivered again once the lease ends.
func (s *Service) record(ctx context.Context, logger *slog.Logger, what string, write func(context.Context) (int64, error)) {
	attempt, cancel := jobs.FinalWriteContext(ctx)
	defer cancel()
	n, err := write(attempt)
	if err != nil {
		logger.ErrorContext(context.WithoutCancel(ctx), what+" failed; it is delivered again when its lease ends", "error", err)
		return
	}
	if n != 1 {
		logger.ErrorContext(context.WithoutCancel(ctx), what, "error", "notification is no longer leased to this delivery")
	}
}

// retryDelay is the wait after the attempt-th failed delivery: retryFirst,
// doubling with each failure, at most retryMax.
func retryDelay(attempt int64) time.Duration {
	d := retryFirst
	for i := int64(1); i < attempt && d < retryMax; i++ {
		d *= 2
	}
	return min(d, retryMax)
}