tinyrelayfile
tinyrelay/internal/daemon/webpush.go/
internal/daemon/webpush.go raw
1package daemon
2
3// Web Push delivers relay notifications to installed browsers. A member
4// registers a device subscription with a signed request; the relay keeps one
5// VAPID key pair per tenant and sends each notification summary encrypted to
6// every registered device. Push services see only ciphertext.
7
8import (
9 "context"
10 "database/sql"
11 "encoding/json"
12 "errors"
13 "fmt"
14 "net/http"
15 "strings"
16 "time"
17
18 "github.com/FelineStateMachine/tinyrelay/internal/relay"
19 "github.com/FelineStateMachine/tinyrelay/internal/replication"
20 "github.com/FelineStateMachine/tinyrelay/internal/storage"
21 "github.com/FelineStateMachine/tinyrelay/internal/webpush"
22 "github.com/FelineStateMachine/tinyrelay/internal/work"
23 "github.com/google/uuid"
24)
25
26const (
27 pushSettingKey = "webpush.vapid"
28 pushMaxPerPubkey = 8
29 pushMaxFailures = 5
30 pushTTL = 24 * time.Hour
31 pushSummaryMaxLength = 160
32)
33
34const pushSchema = `CREATE TABLE IF NOT EXISTS web_push(endpoint TEXT PRIMARY KEY, pubkey TEXT NOT NULL, subscription TEXT NOT NULL, created_at INTEGER NOT NULL, failures INTEGER NOT NULL DEFAULT 0, categories TEXT NOT NULL DEFAULT '');
35CREATE INDEX IF NOT EXISTS web_push_pubkey ON web_push(pubkey);`
36
37type pushPayload struct {
38 Recipient string `json:"recipient"`
39 // Kind is a device category for event summaries or a relay notice kind.
40 Kind string `json:"kind"`
41 Subject string `json:"subject"`
42 Text string `json:"text"`
43 URL string `json:"url,omitempty"`
44 // Actions are the answers the device offers on the notification.
45 Actions []pushAction `json:"actions,omitempty"`
46}
47
48// pushAction is one button on a notification. The service worker opens the
49// notification's url with answer=<action> when it is pressed.
50type pushAction struct {
51 Action string `json:"action"`
52 Title string `json:"title"`
53}
54
55// pushMessage is what the service worker receives after decryption.
56type pushMessage struct {
57 Title string `json:"title"`
58 Body string `json:"body"`
59 URL string `json:"url"`
60 Tag string `json:"tag"`
61 Badge int `json:"badge"`
62 Actions []pushAction `json:"actions,omitempty"`
63}
64
65// pushRegistration is the browser's subscription with its chosen categories.
66type pushRegistration struct {
67 Subscription webpush.Subscription `json:"subscription"`
68 Categories []string `json:"categories"`
69}
70
71func (t *Tenant) initPush(ctx context.Context) error {
72 if _, err := t.store.DB().ExecContext(ctx, pushSchema); err != nil {
73 return err
74 }
75 // Devices registered before categories existed receive everything.
76 _, err := t.store.DB().ExecContext(ctx, "ALTER TABLE web_push ADD COLUMN categories TEXT NOT NULL DEFAULT ''")
77 if err != nil && !strings.Contains(err.Error(), "duplicate column") {
78 return err
79 }
80 return nil
81}
82
83// pushVAPIDKeys returns the tenant's VAPID pair, creating it on first use.
84func (t *Tenant) pushVAPIDKeys(ctx context.Context) (webpush.Keys, error) {
85 t.pushMu.Lock()
86 defer t.pushMu.Unlock()
87 if t.pushVAPID != nil {
88 return *t.pushVAPID, nil
89 }
90 var encoded string
91 err := t.store.GetSetting(ctx, pushSettingKey, &encoded)
92 if err != nil && !errors.Is(err, sql.ErrNoRows) {
93 return webpush.Keys{}, err
94 }
95 var keys webpush.Keys
96 if encoded != "" {
97 keys, err = webpush.ParseKeys(encoded)
98 } else {
99 keys, err = webpush.GenerateKeys()
100 if err == nil {
101 encoded, err = keys.Marshal()
102 }
103 if err == nil {
104 err = t.store.PutSetting(ctx, pushSettingKey, encoded)
105 }
106 }
107 if err != nil {
108 return webpush.Keys{}, err
109 }
110 t.pushVAPID = &keys
111 return keys, nil
112}
113
114// pushSubject identifies the relay operator to push services.
115func (t *Tenant) pushSubject() string {
116 if contact := strings.TrimSpace(t.Policy().Contact); contact != "" && strings.Contains(contact, "@") && !strings.Contains(contact, ":") {
117 return "mailto:" + contact
118 }
119 return t.publicURL
120}
121
122// pushHTTP serves the device notification endpoints under one named
123// operation so the diagnostics registry and traces see them as a unit.
124func (t *Tenant) pushHTTP(w http.ResponseWriter, r *http.Request) {
125 ctx, finish := t.app.telemetry.Start(r.Context(), "push")
126 r = r.WithContext(ctx)
127 outcome := "error"
128 defer func() { finish(outcome) }()
129 outcome = t.servePush(w, r)
130}
131
132// servePush returns the telemetry outcome for the request it answered.
133func (t *Tenant) servePush(w http.ResponseWriter, r *http.Request) string {
134 switch {
135 case r.URL.Path == "/push/key" && r.Method == http.MethodGet:
136 keys, err := t.pushVAPIDKeys(r.Context())
137 if err != nil {
138 writeJSON(w, 500, map[string]string{"error": "push keys unavailable"})
139 return "error"
140 }
141 writeJSON(w, 200, map[string]string{"key": keys.PublicKey()})
142 return "ok"
143 case r.URL.Path == "/inbox/seen" && r.Method == http.MethodPost:
144 // Opening the inbox clears the badge. This is the person's own
145 // read marker, so the browser session may set it from this origin.
146 actor, err := t.resolveUIActor(r)
147 if err != nil || actor == "" {
148 if s, sessionErr := t.cookieReadSession(r, relay.Session{RelayURL: t.RelayURL()}); sessionErr == nil {
149 actor = first(s.PubKeys)
150 }
151 }
152 if actor == "" {
153 writeJSON(w, 401, map[string]string{"error": "auth-required: sign this request"})
154 return "unauthorized"
155 }
156 if err := t.markInboxSeen(r.Context(), actor); err != nil {
157 writeJSON(w, 500, map[string]string{"error": "could not record inbox visit"})
158 return "error"
159 }
160 writeJSON(w, 200, map[string]any{"ok": true})
161 return "ok"
162 case (r.URL.Path == "/push/subscribe" || r.URL.Path == "/push/unsubscribe") && r.Method == http.MethodPost:
163 // Registration is a signed mutation like every other change made
164 // from the browser; the session cookie alone is not enough.
165 actor, err := t.resolveUIActor(r)
166 if err != nil || actor == "" {
167 writeJSON(w, 401, map[string]string{"error": "auth-required: sign this request"})
168 return "unauthorized"
169 }
170 body, err := t.requestBody(w, r)
171 if err != nil {
172 writeJSON(w, 400, map[string]string{"error": err.Error()})
173 return "invalid"
174 }
175 registration, err := parsePushRegistration(body)
176 if err != nil {
177 writeJSON(w, 400, map[string]string{"error": "invalid: push subscription"})
178 return "invalid"
179 }
180 subscription := registration.Subscription
181 if r.URL.Path == "/push/unsubscribe" {
182 if _, err := t.store.DB().ExecContext(r.Context(), "DELETE FROM web_push WHERE endpoint=? AND pubkey=?", subscription.Endpoint, actor); err != nil {
183 writeJSON(w, 500, map[string]string{"error": "unsubscribe failed"})
184 return "error"
185 }
186 writeJSON(w, 200, map[string]any{"ok": true})
187 return "ok"
188 }
189 if role, err := t.community.Role(r.Context(), actor); err != nil || (role == "" && actor != t.Policy().Owner) {
190 writeJSON(w, 403, map[string]string{"error": "restricted: notifications are limited to relay members"})
191 return "unauthorized"
192 }
193 if err := t.savePushSubscription(r.Context(), actor, registration); err != nil {
194 writeJSON(w, statusForPush(err), map[string]string{"error": err.Error()})
195 if statusForPush(err) == 400 {
196 return "invalid"
197 }
198 return "error"
199 }
200 writeJSON(w, 200, map[string]any{"ok": true})
201 return "ok"
202 default:
203 http.NotFound(w, r)
204 return "invalid"
205 }
206}
207
208func statusForPush(err error) int {
209 if strings.HasPrefix(err.Error(), "invalid:") {
210 return 400
211 }
212 return 500
213}
214
215// parsePushRegistration accepts the browser's registration, with or without
216// categories, and validates the subscription inside it.
217func parsePushRegistration(body []byte) (pushRegistration, error) {
218 var registration pushRegistration
219 if err := json.Unmarshal(body, &registration); err != nil || registration.Subscription.Endpoint == "" {
220 registration = pushRegistration{}
221 if err := json.Unmarshal(body, &registration.Subscription); err != nil {
222 return registration, err
223 }
224 }
225 if err := registration.Subscription.Validate(); err != nil {
226 return registration, err
227 }
228 var kept []string
229 for _, category := range registration.Categories {
230 if containsString(pushCategories, category) && !containsString(kept, category) {
231 kept = append(kept, category)
232 }
233 }
234 registration.Categories = kept
235 return registration, nil
236}
237
238func (t *Tenant) savePushSubscription(ctx context.Context, actor string, registration pushRegistration) error {
239 subscription := registration.Subscription
240 raw, err := json.Marshal(subscription)
241 if err != nil {
242 return err
243 }
244 categories := ""
245 if len(registration.Categories) > 0 {
246 encoded, err := json.Marshal(registration.Categories)
247 if err != nil {
248 return err
249 }
250 categories = string(encoded)
251 }
252 return t.store.WithTx(ctx, func(tx *sql.Tx) error {
253 var count int
254 if err := tx.QueryRowContext(ctx, "SELECT count(*) FROM web_push WHERE pubkey=? AND endpoint<>?", actor, subscription.Endpoint).Scan(&count); err != nil {
255 return err
256 }
257 if count >= pushMaxPerPubkey {
258 return fmt.Errorf("invalid: at most %d devices per key", pushMaxPerPubkey)
259 }
260 _, err := tx.ExecContext(ctx, "INSERT INTO web_push(endpoint,pubkey,subscription,created_at,failures,categories) VALUES(?,?,?,?,0,?) ON CONFLICT(endpoint) DO UPDATE SET pubkey=excluded.pubkey, subscription=excluded.subscription, failures=0, categories=excluded.categories", subscription.Endpoint, actor, string(raw), time.Now().Unix(), categories)
261 return err
262 })
263}
264
265// enqueuePush records one notification summary for device delivery. It is
266// suitable for records.Config.PushNotification and does nothing for
267// recipients without registered devices.
268func (t *Tenant) enqueuePush(ctx context.Context, recipient, kind, subject, text string) error {
269 return t.enqueuePushPayload(ctx, pushPayload{Recipient: recipient, Kind: kind, Subject: subject, Text: text})
270}
271
272func (t *Tenant) enqueuePushPayload(ctx context.Context, payload pushPayload) error {
273 var count int
274 if err := t.store.DB().QueryRowContext(ctx, "SELECT count(*) FROM web_push WHERE pubkey=?", payload.Recipient).Scan(&count); err != nil || count == 0 {
275 return err
276 }
277 if len(payload.Text) > pushSummaryMaxLength {
278 payload.Text = payload.Text[:pushSummaryMaxLength]
279 }
280 recipient := payload.Recipient
281 encoded, err := json.Marshal(payload)
282 if err != nil {
283 return err
284 }
285 payloadJSON := string(encoded)
286 return t.store.WithTx(ctx, func(tx *sql.Tx) error {
287 return storage.AddIntents(ctx, tx, []storage.Intent{{Kind: notificationPush, EventID: uuid.NewString(), Target: recipient, Payload: payloadJSON}}, time.Now().Unix())
288 })
289}
290
291func (t *Tenant) handleNotificationPush(ctx context.Context, intent work.Intent) error {
292 var payload pushPayload
293 if err := json.Unmarshal([]byte(intent.Payload), &payload); err != nil || payload.Recipient != intent.Target {
294 return errors.New("notification-push: invalid payload")
295 }
296 rows, err := t.store.DB().QueryContext(ctx, "SELECT endpoint, subscription, failures, categories FROM web_push WHERE pubkey=?", payload.Recipient)
297 if err != nil {
298 return err
299 }
300 type device struct {
301 endpoint string
302 subscription webpush.Subscription
303 failures int
304 }
305 category := payload.Kind
306 if !containsString(pushCategories, category) {
307 category = pushRelay
308 }
309 var devices []device
310 for rows.Next() {
311 var item device
312 var raw, categories string
313 if err := rows.Scan(&item.endpoint, &raw, &item.failures, &categories); err != nil {
314 rows.Close()
315 return err
316 }
317 // A device that chose categories only wakes for those.
318 if allowed := decodeCategories(categories); allowed != nil && !allowed[category] {
319 continue
320 }
321 if json.Unmarshal([]byte(raw), &item.subscription) == nil {
322 devices = append(devices, item)
323 }
324 }
325 rows.Close()
326 if len(devices) == 0 {
327 return nil
328 }
329 keys, err := t.pushVAPIDKeys(ctx)
330 if err != nil {
331 return err
332 }
333 message, err := json.Marshal(t.pushMessage(ctx, payload))
334 if err != nil {
335 return err
336 }
337 client := t.pushClient
338 if client == nil {
339 client = replication.NewPinnedClient(10 * time.Second)
340 }
341 var result error
342 delivered, removed, failed := 0, 0, 0
343 for _, item := range devices {
344 sent, sendErr := webpush.Send(ctx, client, keys, t.pushSubject(), item.subscription, message, pushTTL)
345 switch {
346 case sendErr == nil:
347 delivered++
348 _, _ = t.store.DB().ExecContext(ctx, "UPDATE web_push SET failures=0 WHERE endpoint=?", item.endpoint)
349 case sent.Gone || item.failures+1 >= pushMaxFailures:
350 removed++
351 _, _ = t.store.DB().ExecContext(ctx, "DELETE FROM web_push WHERE endpoint=?", item.endpoint)
352 default:
353 failed++
354 _, _ = t.store.DB().ExecContext(ctx, "UPDATE web_push SET failures=failures+1 WHERE endpoint=?", item.endpoint)
355 result = errors.Join(result, sendErr)
356 }
357 }
358 // Counts only: endpoints and keys stay out of logs, as they do for metrics.
359 t.app.telemetry.Logger().Info("device notifications sent", "tenant", t.meta.Name, "category", category, "delivered", delivered, "failed", failed, "removed", removed)
360 return result
361}
362
363// pushMessage turns a notification summary into what the device shows.
364func (t *Tenant) pushMessage(ctx context.Context, payload pushPayload) pushMessage {
365 title := t.Policy().Name
366 if title == "" {
367 title = t.meta.Name
368 }
369 body := payload.Text
370 if body == "" {
371 body = map[string]string{
372 "reports": "A new report needs moderation.", "jobs": "A background job finished.",
373 "succession": "Relay succession changed.", "digest": "Your relay digest is ready.",
374 "test": "This is a test notification.", "succession-transfer": "Relay ownership was transferred.",
375 }[payload.Kind]
376 }
377 if body == "" {
378 body = "New relay notification."
379 }
380 if payload.Subject != "" && payload.Subject != "relay" && !strings.EqualFold(payload.Subject, payload.Kind) {
381 title = title + ": " + payload.Subject
382 }
383 url := payload.URL
384 if url == "" {
385 url = strings.TrimRight(t.publicURL, "/") + "/chat"
386 }
387 return pushMessage{Title: title, Body: body, URL: url, Tag: "tiny-" + payload.Kind, Badge: t.inboxUnread(ctx, payload.Recipient), Actions: payload.Actions}
388}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs