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
	}
	now := time.Now().Unix()
	// 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
		}
		if derr := storage.TouchPushSubscription(sub.Endpoint, now); 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. The blob mirrors localStorage: a JSON object
// whose "pete.disabledSources.v1" value is itself a JSON string encoding a
// {sourceName: true} map. Any parse failure yields an empty (deny-nothing) set.
func disabledSourcesFor(sub string) map[string]bool {
	out := map[string]bool{}
	blob, err := storage.GetUserPrefs(sub)
	if err != nil || blob == "" {
		return out
	}
	var prefs map[string]json.RawMessage
	if err := json.Unmarshal([]byte(blob), &prefs); err != nil {
		return out
	}
	raw, ok := prefs["pete.disabledSources.v1"]
	if !ok {
		return out
	}
	// The value is normally a JSON *string* containing JSON; unwrap that first,
	// but tolerate a bare object too.
	inner := []byte(raw)
	var asStr string
	if err := json.Unmarshal(raw, &asStr); err == nil {
		inner = []byte(asStr)
	}
	var set map[string]bool
	if err := json.Unmarshal(inner, &set); err != nil {
		return out
	}
	for name, on := range set {
		if on {
			out[name] = true
		}
	}
	return out
}

// 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)
}