Pete/internal/web/push_sender.go

package web

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"time"

	"pete/internal/safehttp"
	"pete/internal/storage"

	webpush "github.com/SherClockHolmes/webpush-go"
)

// digestScan caps how many new stories the sender inspects per subscriber in one
// pass. Well past MinStories; the digest only needs a count and one headline.
const digestScan = 60

// pushSendTimeout bounds one push delivery. The endpoint is user-supplied, so a
// hostile or dead push service must not be able to wedge the (serial) digest
// loop and starve every other subscriber.
const pushSendTimeout = 15 * time.Second

// runPushSender periodically builds and delivers a "N new stories" digest to
// each subscriber, respecting their disabled-sources preference. It's started
// only when push is configured. Best-effort throughout: a failed send never
// stops the loop, and a permanently-gone endpoint is pruned.
func (s *Server) runPushSender(ctx context.Context) {
	interval := time.Duration(s.cfg.Push.IntervalMinutes) * time.Minute
	if interval <= 0 {
		interval = 6 * time.Hour
	}
	slog.Info("web: push digest sender started", "interval", interval, "min_stories", s.cfg.Push.MinStories)
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.sendDigests()
		}
	}
}

// sendDigests walks every subscription once, notifying those with enough new
// stories since their last digest.
func (s *Server) sendDigests() {
	subs, err := storage.ListPushSubscriptions()
	if err != nil {
		slog.Error("push: list subscriptions failed", "err", err)
		return
	}
	if len(subs) == 0 {
		return
	}
	// Disabled-source sets are per user; cache within a pass so a user with
	// several devices only parses prefs once.
	disabledByUser := make(map[string]map[string]bool)
	sent, pruned := 0, 0

	for _, sub := range subs {
		stories, err := storage.NewClassifiedSince(sub.LastNotifiedAt, digestScan)
		if err != nil {
			slog.Error("push: scan new stories failed", "sub", sub.UserSub, "err", err)
			continue
		}
		if len(stories) == 0 {
			continue
		}
		disabled, ok := disabledByUser[sub.UserSub]
		if !ok {
			disabled = disabledSourcesFor(sub.UserSub)
			disabledByUser[sub.UserSub] = disabled
		}
		count, top := 0, ""
		for _, st := range stories {
			if disabled[st.Source] {
				continue
			}
			if count == 0 {
				top = st.Headline
			}
			count++
		}
		if count < s.cfg.Push.MinStories {
			// If a full scan window filled entirely with stories the user has
			// hidden, the non-hidden count can stay below the threshold forever
			// while the watermark never advances — the same hidden window is
			// re-scanned every pass and the subscriber is permanently starved of
			// digests. When the window was capped (a genuine glut, not a quiet
			// spell), step the watermark past it so the next pass sees fresh
			// stories. A non-full window is a real lull; leave it to accumulate.
			if len(stories) == digestScan {
				if derr := storage.TouchPushSubscription(sub.Endpoint, stories[0].SeenAt); derr != nil {
					slog.Error("push: advance watermark past hidden glut failed", "err", derr)
				}
			}
			continue
		}

		payload := buildDigestPayload(count, top)
		gone, err := s.sendPush(sub, payload)
		if gone {
			if derr := storage.RemovePushSubscription(sub.Endpoint); derr != nil {
				slog.Error("push: prune gone subscription failed", "err", derr)
			} else {
				pruned++
			}
			continue
		}
		if err != nil {
			slog.Warn("push: send failed", "sub", sub.UserSub, "err", err)
			continue
		}
		// Advance to the newest story this digest actually accounted for, not a
		// pass-start "now": the loop can run for minutes (one slow endpoint per
		// send), so stories arriving mid-pass would otherwise be re-counted next
		// pass. stories[0] is the newest in the scanned window (seen_at DESC).
		if derr := storage.TouchPushSubscription(sub.Endpoint, stories[0].SeenAt); derr != nil {
			slog.Error("push: advance watermark failed", "err", derr)
		}
		sent++
	}
	if sent > 0 || pruned > 0 {
		slog.Info("push: digest pass complete", "sent", sent, "pruned", pruned, "subscriptions", len(subs))
	}
}

// buildDigestPayload renders the notification JSON the service worker expects.
func buildDigestPayload(count int, top string) []byte {
	body := fmt.Sprintf("%d new stories", count)
	if count == 1 {
		body = "1 new story"
	}
	if top != "" {
		body += ": " + top
	}
	b, _ := json.Marshal(map[string]string{
		"title": "Pete has fresh news",
		"body":  body,
		"url":   "/",
		"tag":   "pete-digest",
	})
	return b
}

// sendPush encrypts and delivers one notification. It reports gone=true when the
// push service says the endpoint no longer exists (404/410) so the caller can
// prune it; err is set for other, likely-transient failures.
func (s *Server) sendPush(sub storage.PushSubscription, payload []byte) (gone bool, err error) {
	resp, err := webpush.SendNotification(payload, &webpush.Subscription{
		Endpoint: sub.Endpoint,
		Keys:     webpush.Keys{P256dh: sub.P256dh, Auth: sub.Auth},
	}, &webpush.Options{
		// The endpoint URL comes from the browser and is attacker-influenceable,
		// so deliver through the SSRF-guarded client (blocks loopback/RFC1918/
		// link-local/metadata targets) with a hard timeout — never the library's
		// unguarded default http.Client.
		HTTPClient:      s.pushClient(),
		Subscriber:      s.cfg.Push.Subject,
		VAPIDPublicKey:  s.cfg.Push.VAPIDPublicKey,
		VAPIDPrivateKey: s.cfg.Push.VAPIDPrivateKey,
		TTL:             24 * 60 * 60,
		Urgency:         webpush.UrgencyNormal,
	})
	if err != nil {
		return false, err
	}
	defer resp.Body.Close()
	if resp.StatusCode == 404 || resp.StatusCode == 410 {
		return true, nil
	}
	if resp.StatusCode >= 400 {
		return false, fmt.Errorf("push service returned %d", resp.StatusCode)
	}
	return false, nil
}

// disabledSourcesFor returns the set of source names a user has hidden, read
// from their stored prefs blob. Any parse failure yields an empty (deny-nothing)
// set — see userPrefBoolSet for the blob's shape and why it is double-encoded.
func disabledSourcesFor(sub string) map[string]bool {
	return userPrefBoolSet(sub, "pete.disabledSources.v1")
}

// pushClient returns the shared SSRF-guarded, timeout-bounded HTTP client used
// for Web Push delivery, building it once on first use. The digest loop is a
// single goroutine, so the lazy init needs no lock.
func (s *Server) pushClient() *http.Client {
	if s.pushHTTP == nil {
		s.pushHTTP = safehttp.NewClient(pushSendTimeout)
	}
	return s.pushHTTP
}

// StartPushSender launches the digest loop if push is enabled. Safe to call
// unconditionally; it's a no-op when push is off.
func (s *Server) StartPushSender(ctx context.Context) {
	if !s.cfg.Push.Enabled || s.auth == nil {
		return
	}
	go s.runPushSender(ctx)
}