byre/internal/push/push.go

package push

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

	"github.com/SherClockHolmes/webpush-go"

	"g.fc.run/misc/byre/internal/api"
	"g.fc.run/misc/byre/internal/store"
)

type Message struct {
	Title string `json:"title"`
	Body  string `json:"body"`
	URL   string `json:"url"`
}

type Notifier struct {
	store  *store.Store
	keys   store.VAPIDKeys
	window time.Duration
	items  chan api.InboxItem
	done   chan struct{}
	exited chan struct{}
}

func New(st *store.Store, window time.Duration) (*Notifier, error) {
	ctx := context.Background()
	keys, err := st.GetVAPIDKeys(ctx)
	if errors.Is(err, store.ErrNotFound) {
		priv, pub, err := webpush.GenerateVAPIDKeys()
		if err != nil {
			return nil, err
		}
		keys, err = st.InitVAPIDKeys(ctx, store.VAPIDKeys{Public: pub, Private: priv})
		if err != nil {
			return nil, err
		}
	} else if err != nil {
		return nil, err
	}

	n := &Notifier{
		store:  st,
		keys:   keys,
		window: window,
		items:  make(chan api.InboxItem, 256),
		done:   make(chan struct{}),
		exited: make(chan struct{}),
	}
	go n.run()
	return n, nil
}

func (n *Notifier) PublicKey() string {
	return n.keys.Public
}

func (n *Notifier) Close() {
	close(n.done)
	<-n.exited
}

func (n *Notifier) Observe(ev api.Event) {
	if ev.Kind != "inbox.created" {
		return
	}

	b, err := json.Marshal(ev.Payload)
	if err != nil {
		slog.Error("push: encode inbox item", "err", err)
		return
	}
	item := api.InboxItem{}
	err = json.Unmarshal(b, &item)
	if err != nil {
		slog.Error("push: decode inbox item", "err", err)
		return
	}

	select {
	case n.items <- item:
	default:
		slog.Warn("push: queue full, dropping notification", "item", item.Id)
	}
}

func (n *Notifier) run() {
	defer close(n.exited)

	pending := []api.InboxItem{}
	var flush <-chan time.Time
	for {
		select {
		case item := <-n.items:
			pending = append(pending, item)
			if flush == nil {
				flush = time.After(n.window)
			}
		case <-flush:
			n.send(pending)
			pending = []api.InboxItem{}
			flush = nil
		case <-n.done:
			return
		}
	}
}

func Wants(p api.NotificationPrefs, item api.InboxItem) bool {
	switch item.Kind {
	case api.Question:
		if item.Blocking {
			return p.BlockingQuestions
		}
		return p.Questions
	case api.Report:
		return p.Reports
	case api.SessionError:
		return p.Errors
	case api.RateLimited:
		return p.RateLimits
	}
	return false
}

func (n *Notifier) send(items []api.InboxItem) {
	ctx := context.Background()
	targets, err := n.store.PushTargets(ctx)
	if err != nil {
		slog.Error("push: list subscriptions", "err", err)
		return
	}

	titles := map[string]string{}
	for _, item := range items {
		if _, ok := titles[item.TaskId]; ok {
			continue
		}
		t, err := n.store.GetTask(ctx, item.TaskId)
		if err != nil {
			slog.Error("push: get task", "task", item.TaskId, "err", err)
			return
		}
		titles[item.TaskId] = t.Title
	}

	for _, target := range targets {
		wanted := []api.InboxItem{}
		for _, item := range items {
			if Wants(target.Prefs, item) {
				wanted = append(wanted, item)
			}
		}
		if len(wanted) == 0 {
			continue
		}
		n.deliver(ctx, target.Subscription, Compose(wanted, titles), urgency(wanted))
	}
}

func urgency(items []api.InboxItem) webpush.Urgency {
	for _, item := range items {
		if item.Blocking {
			return webpush.UrgencyHigh
		}
	}
	return webpush.UrgencyNormal
}

func Compose(items []api.InboxItem, titles map[string]string) Message {
	first := items[0]
	url := "/#/task/" + first.TaskId
	if len(items) == 1 {
		return Message{Title: "byre · " + titles[first.TaskId], Body: itemText(first), URL: url}
	}

	lines := []string{}
	for _, item := range items {
		lines = append(lines, titles[item.TaskId]+": "+itemText(item))
	}
	return Message{Title: fmt.Sprintf("byre · %d new inbox items", len(items)), Body: strings.Join(lines, "\n"), URL: url}
}

func itemText(item api.InboxItem) string {
	text := ""
	switch item.Kind {
	case api.Question:
		text, _ = item.Payload["question"].(string)
	case api.Report:
		text, _ = item.Payload["text"].(string)
	case api.SessionError:
		e, _ := item.Payload["error"].(string)
		text = "Error: " + e
	case api.RateLimited:
		text = "Rate limited"
	}
	text, _, _ = strings.Cut(strings.TrimSpace(text), "\n")
	runes := []rune(text)
	if len(runes) > 200 {
		text = string(runes[:199]) + "…"
	}
	return text
}

func (n *Notifier) deliver(ctx context.Context, sub store.PushSubscription, msg Message, u webpush.Urgency) {
	payload, err := json.Marshal(msg)
	if err != nil {
		slog.Error("push: encode message", "err", err)
		return
	}

	ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
	defer cancel()
	resp, err := webpush.SendNotificationWithContext(ctx, payload, &webpush.Subscription{
		Endpoint: sub.Endpoint,
		Keys:     webpush.Keys{P256dh: sub.P256dh, Auth: sub.Auth},
	}, &webpush.Options{
		Subscriber:      sub.Subscriber,
		TTL:             24 * 60 * 60,
		Urgency:         u,
		VAPIDPublicKey:  n.keys.Public,
		VAPIDPrivateKey: n.keys.Private,
	})
	if err != nil {
		slog.Warn("push: send", "user", sub.UserLogin, "err", err)
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone {
		slog.Info("push: subscription expired", "user", sub.UserLogin, "status", resp.StatusCode)
		err = n.store.RemovePushSubscription(ctx, sub.Endpoint, sub.UserLogin, "byre")
		if err != nil {
			slog.Error("push: remove subscription", "err", err)
		}
		return
	}
	if resp.StatusCode >= 300 {
		body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
		slog.Warn("push: rejected", "user", sub.UserLogin, "status", resp.StatusCode, "body", string(body))
	}
}