Add web push notifications for new inbox items
This commit is contained in:
Ian Gulliver
@@ -0,0 +1,251 @@
|
||||
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))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,170 @@
|
||||
package push_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/ecdh"
|
||||
"crypto/rand"
|
||||
"encoding/base64"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"g.fc.run/misc/byre/internal/api"
|
||||
"g.fc.run/misc/byre/internal/push"
|
||||
"g.fc.run/misc/byre/internal/store"
|
||||
)
|
||||
|
||||
type received struct {
|
||||
urgency string
|
||||
auth string
|
||||
}
|
||||
|
||||
func TestBurstIsOneNotification(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
var notifier *push.Notifier
|
||||
st, err := store.Open(filepath.Join(t.TempDir(), "byre.db"), func(ev api.Event) { notifier.Observe(ev) })
|
||||
if err != nil {
|
||||
t.Fatalf("open store: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { st.Close() })
|
||||
|
||||
notifier, err = push.New(st, 100*time.Millisecond)
|
||||
if err != nil {
|
||||
t.Fatalf("new notifier: %v", err)
|
||||
}
|
||||
t.Cleanup(notifier.Close)
|
||||
|
||||
got := make(chan received, 10)
|
||||
endpoint := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
got <- received{urgency: r.Header.Get("Urgency"), auth: r.Header.Get("Authorization")}
|
||||
w.WriteHeader(http.StatusCreated)
|
||||
}))
|
||||
t.Cleanup(endpoint.Close)
|
||||
gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusGone)
|
||||
}))
|
||||
t.Cleanup(gone.Close)
|
||||
|
||||
now := time.Now().UTC()
|
||||
_, err = st.EnsureUser(ctx, "alice", "Alice")
|
||||
if err != nil {
|
||||
t.Fatalf("ensure user: %v", err)
|
||||
}
|
||||
err = st.CreateProject(ctx, api.Project{Id: "p1", Name: "demo", RepoPath: "/nonexistent", DefaultBranch: "main", Remote: "none", CreatedBy: "alice", CreatedAt: now})
|
||||
if err != nil {
|
||||
t.Fatalf("create project: %v", err)
|
||||
}
|
||||
err = st.CreateTask(ctx, api.Task{Id: "t1", ProjectId: "p1", Title: "Fix login", Branch: "b", WorktreePath: "/nonexistent", BaseRef: "x", Status: api.TaskStatusActive, CreatedBy: "alice", CreatedAt: now})
|
||||
if err != nil {
|
||||
t.Fatalf("create task: %v", err)
|
||||
}
|
||||
err = st.CreateSession(ctx, api.AgentSession{Id: "s1", TaskId: "t1", Vendor: api.Claude, VendorSessionId: "s1", State: api.AgentSessionStateIdle, Role: api.Implementer, CreatedBy: "alice", CreatedAt: now}, "p1")
|
||||
if err != nil {
|
||||
t.Fatalf("create session: %v", err)
|
||||
}
|
||||
|
||||
for _, url := range []string{endpoint.URL, gone.URL} {
|
||||
err = st.AddPushSubscription(ctx, store.PushSubscription{Endpoint: url, UserLogin: "alice", P256dh: clientKey(t), Auth: secret(t), Subscriber: "https://byre.example"})
|
||||
if err != nil {
|
||||
t.Fatalf("add subscription: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
create := func(id string, kind api.InboxItemKind, blocking bool, payload map[string]any) {
|
||||
t.Helper()
|
||||
err := st.CreateInboxItem(ctx, api.InboxItem{Id: id, ProjectId: "p1", TaskId: "t1", AgentSessionId: "s1", ThreadId: id,
|
||||
Kind: kind, Priority: api.Blocked, Blocking: blocking, Payload: payload, State: api.Open, CreatedAt: now})
|
||||
if err != nil {
|
||||
t.Fatalf("create inbox item: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
create("r1", api.Report, false, map[string]any{"text": "done"})
|
||||
expectNone(t, got)
|
||||
|
||||
create("q1", api.Question, true, map[string]any{"question": "Which port?"})
|
||||
create("q2", api.Question, true, map[string]any{"question": "Which host?"})
|
||||
create("r2", api.Report, false, map[string]any{"text": "done"})
|
||||
|
||||
select {
|
||||
case r := <-got:
|
||||
if r.urgency != "high" || !strings.HasPrefix(r.auth, "vapid t=") {
|
||||
t.Fatalf("unexpected push request %+v", r)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatalf("no push delivered")
|
||||
}
|
||||
expectNone(t, got)
|
||||
|
||||
targets, err := st.PushTargets(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("push targets: %v", err)
|
||||
}
|
||||
if len(targets) != 1 || targets[0].Subscription.Endpoint != endpoint.URL {
|
||||
t.Fatalf("expired subscription not removed: %+v", targets)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompose(t *testing.T) {
|
||||
titles := map[string]string{"t1": "Fix login", "t2": "Add search"}
|
||||
one := push.Compose([]api.InboxItem{
|
||||
{TaskId: "t1", Kind: api.Question, Payload: map[string]any{"question": "Which port?\nmore"}},
|
||||
}, titles)
|
||||
if one != (push.Message{Title: "byre · Fix login", Body: "Which port?", URL: "/#/task/t1"}) {
|
||||
t.Fatalf("unexpected single message %+v", one)
|
||||
}
|
||||
|
||||
burst := push.Compose([]api.InboxItem{
|
||||
{TaskId: "t2", Kind: api.Question, Payload: map[string]any{"question": "Which index?"}},
|
||||
{TaskId: "t1", Kind: api.SessionError, Payload: map[string]any{"error": "exit 1"}},
|
||||
}, titles)
|
||||
want := push.Message{Title: "byre · 2 new inbox items", Body: "Add search: Which index?\nFix login: Error: exit 1", URL: "/#/task/t2"}
|
||||
if burst != want {
|
||||
t.Fatalf("unexpected burst message %+v", burst)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWantsDefaults(t *testing.T) {
|
||||
prefs := api.NotificationPrefs{BlockingQuestions: true}
|
||||
cases := []struct {
|
||||
item api.InboxItem
|
||||
want bool
|
||||
}{
|
||||
{api.InboxItem{Kind: api.Question, Blocking: true}, true},
|
||||
{api.InboxItem{Kind: api.Question}, false},
|
||||
{api.InboxItem{Kind: api.Report}, false},
|
||||
{api.InboxItem{Kind: api.SessionError, Blocking: true}, false},
|
||||
{api.InboxItem{Kind: api.RateLimited}, false},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if push.Wants(prefs, c.item) != c.want {
|
||||
t.Fatalf("wants %+v: got %v", c.item, !c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func expectNone(t *testing.T, got chan received) {
|
||||
t.Helper()
|
||||
select {
|
||||
case r := <-got:
|
||||
t.Fatalf("unexpected push %+v", r)
|
||||
case <-time.After(400 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
|
||||
func clientKey(t *testing.T) string {
|
||||
k, err := ecdh.P256().GenerateKey(rand.Reader)
|
||||
if err != nil {
|
||||
t.Fatalf("generate key: %v", err)
|
||||
}
|
||||
return base64.RawURLEncoding.EncodeToString(k.PublicKey().Bytes())
|
||||
}
|
||||
|
||||
func secret(t *testing.T) string {
|
||||
b := make([]byte, 16)
|
||||
rand.Read(b)
|
||||
return base64.RawURLEncoding.EncodeToString(b)
|
||||
}
|
||||
Reference in New Issue
Block a user