spsays-bff/internal/service/push_sender.go

package service

import (
	"context"
	"encoding/json"
	"net/http"
	"strconv"
	"time"

	webpush "github.com/SherClockHolmes/webpush-go"
	"github.com/prabhupadasays/spsays-bff/internal/repository"
	"go.uber.org/zap"
)

const pushMaxAttempts = 3

// PushPayload is the JSON the service worker receives in its `push` event.
type PushPayload struct {
	Title string `json:"title"`
	Body  string `json:"body"`
	URL   string `json:"url,omitempty"`
	Icon  string `json:"icon,omitempty"`
	Tag   string `json:"tag,omitempty"`
	// RequireInteraction keeps the notification on screen until the user acts
	// on it (used for sādhana reminders, which shouldn't auto-dismiss).
	RequireInteraction bool `json:"require_interaction,omitempty"`
}

// PushSender delivers Web Push messages via VAPID. Construct with
// NewPushSender; it returns nil when VAPID isn't configured, and every method
// is nil-safe so callers needn't branch.
type PushSender struct {
	repo    *repository.PushRepo
	pub     string
	priv    string
	subject string
	logger  *zap.Logger
}

// NewPushSender returns nil if the VAPID keys are absent (push disabled).
func NewPushSender(repo *repository.PushRepo, publicKey, privateKey, subject string, logger *zap.Logger) *PushSender {
	if publicKey == "" || privateKey == "" || repo == nil {
		return nil
	}
	return &PushSender{repo: repo, pub: publicKey, priv: privateKey, subject: subject, logger: logger}
}

// PublicKey is the VAPID application server key the browser subscribes with.
func (p *PushSender) PublicKey() string {
	if p == nil {
		return ""
	}
	return p.pub
}

// Enabled reports whether push delivery is configured.
func (p *PushSender) Enabled() bool { return p != nil }

// SendToUser pushes a payload to every subscription the user has, pruning any
// endpoint the push service reports as gone (404/410). Best-effort: failures
// are logged, never returned.
func (p *PushSender) SendToUser(ctx context.Context, userID string, payload PushPayload) {
	if p == nil {
		return
	}
	subs, err := p.repo.ListByUser(ctx, userID)
	if err != nil {
		p.logger.Warn("push: list subscriptions failed", zap.String("user", userID), zap.Error(err))
		return
	}
	if len(subs) == 0 {
		return
	}
	body, err := json.Marshal(payload)
	if err != nil {
		return
	}
	var sent, pruned, failed int
	for _, s := range subs {
		sub := &webpush.Subscription{
			Endpoint: s.Endpoint,
			Keys:     webpush.Keys{P256dh: s.P256dh, Auth: s.Auth},
		}
		status, err := p.sendWithRetry(ctx, body, sub)
		if err != nil {
			failed++
			p.logger.Warn("push: send failed", zap.String("user", userID), zap.Error(err))
			continue
		}
		switch {
		// Subscription expired/gone — drop it so we stop trying.
		case status == http.StatusNotFound || status == http.StatusGone:
			pruned++
			if derr := p.repo.DeleteByEndpoint(ctx, s.Endpoint); derr != nil {
				p.logger.Warn("push: prune failed", zap.Error(derr))
			}
		case status >= 400:
			failed++
			p.logger.Warn("push: non-2xx", zap.Int("status", status), zap.String("user", userID))
		default:
			sent++
		}
	}
	p.logger.Debug("push: delivery summary",
		zap.String("user", userID),
		zap.Int("sent", sent), zap.Int("pruned", pruned), zap.Int("failed", failed))
}

// sendWithRetry posts one notification, retrying on transient failures
// (network error or 429/5xx) with a short backoff that honors Retry-After.
// Returns the final HTTP status, or an error if every attempt errored.
func (p *PushSender) sendWithRetry(ctx context.Context, body []byte, sub *webpush.Subscription) (int, error) {
	var lastErr error
	for attempt := 1; attempt <= pushMaxAttempts; attempt++ {
		resp, err := webpush.SendNotification(body, sub, &webpush.Options{
			Subscriber:      p.subject,
			VAPIDPublicKey:  p.pub,
			VAPIDPrivateKey: p.priv,
			// Hold the message for up to 6h so a sleeping/offline device
			// still gets it on wake (60s would silently expire meanwhile).
			TTL:     6 * 60 * 60,
			Urgency: webpush.UrgencyNormal,
		})
		if err != nil {
			lastErr = err
			if attempt < pushMaxAttempts {
				sleepCtx(ctx, backoff(attempt))
				continue
			}
			return 0, err
		}
		status := resp.StatusCode
		retryAfter := resp.Header.Get("Retry-After")
		_ = resp.Body.Close()
		if isTransientStatus(status) && attempt < pushMaxAttempts {
			sleepCtx(ctx, retryDelay(retryAfter, attempt))
			continue
		}
		return status, nil
	}
	return 0, lastErr
}

func isTransientStatus(s int) bool {
	switch s {
	case http.StatusTooManyRequests, // 429
		http.StatusInternalServerError, // 500
		http.StatusBadGateway,          // 502
		http.StatusServiceUnavailable,  // 503
		http.StatusGatewayTimeout:      // 504
		return true
	}
	return false
}

func backoff(attempt int) time.Duration {
	return time.Duration(attempt) * 500 * time.Millisecond
}

// retryDelay prefers the push service's Retry-After (seconds), capped, else a
// plain backoff.
func retryDelay(retryAfter string, attempt int) time.Duration {
	if retryAfter != "" {
		if secs, err := strconv.Atoi(retryAfter); err == nil && secs > 0 {
			d := time.Duration(secs) * time.Second
			if d > 30*time.Second {
				d = 30 * time.Second
			}
			return d
		}
	}
	return backoff(attempt)
}

// sleepCtx waits for d, returning early if the context is cancelled.
func sleepCtx(ctx context.Context, d time.Duration) {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
	case <-t.C:
	}
}