1package daemon23// Web Push delivers relay notifications to installed browsers. A member4// registers a device subscription with a signed request; the relay keeps one5// VAPID key pair per tenant and sends each notification summary encrypted to6// every registered device. Push services see only ciphertext.78import (9 "context"10 "database/sql"11 "encoding/json"12 "errors"13 "fmt"14 "net/http"15 "strings"16 "time"1718 "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)2526const (27 pushSettingKey = "webpush.vapid"28 pushMaxPerPubkey = 829 pushMaxFailures = 530 pushTTL = 24 * time.Hour31 pushSummaryMaxLength = 16032)3334const 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);`3637type 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}4748// pushAction is one button on a notification. The service worker opens the49// notification's url with answer=<action> when it is pressed.50type pushAction struct {51 Action string `json:"action"`52 Title string `json:"title"`53}5455// 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}6465// 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}7071func (t *Tenant) initPush(ctx context.Context) error {72 if _, err := t.store.DB().ExecContext(ctx, pushSchema); err != nil {73 return err74 }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 err79 }80 return nil81}8283// 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, nil89 }90 var encoded string91 err := t.store.GetSetting(ctx, pushSettingKey, &encoded)92 if err != nil && !errors.Is(err, sql.ErrNoRows) {93 return webpush.Keys{}, err94 }95 var keys webpush.Keys96 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{}, err109 }110 t.pushVAPID = &keys111 return keys, nil112}113114// 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:" + contact118 }119 return t.publicURL120}121122// pushHTTP serves the device notification endpoints under one named123// 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}131132// 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 own145 // 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 made164 // 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.Subscription181 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}207208func statusForPush(err error) int {209 if strings.HasPrefix(err.Error(), "invalid:") {210 return 400211 }212 return 500213}214215// parsePushRegistration accepts the browser's registration, with or without216// categories, and validates the subscription inside it.217func parsePushRegistration(body []byte) (pushRegistration, error) {218 var registration pushRegistration219 if err := json.Unmarshal(body, ®istration); err != nil || registration.Subscription.Endpoint == "" {220 registration = pushRegistration{}221 if err := json.Unmarshal(body, ®istration.Subscription); err != nil {222 return registration, err223 }224 }225 if err := registration.Subscription.Validate(); err != nil {226 return registration, err227 }228 var kept []string229 for _, category := range registration.Categories {230 if containsString(pushCategories, category) && !containsString(kept, category) {231 kept = append(kept, category)232 }233 }234 registration.Categories = kept235 return registration, nil236}237238func (t *Tenant) savePushSubscription(ctx context.Context, actor string, registration pushRegistration) error {239 subscription := registration.Subscription240 raw, err := json.Marshal(subscription)241 if err != nil {242 return err243 }244 categories := ""245 if len(registration.Categories) > 0 {246 encoded, err := json.Marshal(registration.Categories)247 if err != nil {248 return err249 }250 categories = string(encoded)251 }252 return t.store.WithTx(ctx, func(tx *sql.Tx) error {253 var count int254 if err := tx.QueryRowContext(ctx, "SELECT count(*) FROM web_push WHERE pubkey=? AND endpoint<>?", actor, subscription.Endpoint).Scan(&count); err != nil {255 return err256 }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 err262 })263}264265// enqueuePush records one notification summary for device delivery. It is266// suitable for records.Config.PushNotification and does nothing for267// 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}271272func (t *Tenant) enqueuePushPayload(ctx context.Context, payload pushPayload) error {273 var count int274 if err := t.store.DB().QueryRowContext(ctx, "SELECT count(*) FROM web_push WHERE pubkey=?", payload.Recipient).Scan(&count); err != nil || count == 0 {275 return err276 }277 if len(payload.Text) > pushSummaryMaxLength {278 payload.Text = payload.Text[:pushSummaryMaxLength]279 }280 recipient := payload.Recipient281 encoded, err := json.Marshal(payload)282 if err != nil {283 return err284 }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}290291func (t *Tenant) handleNotificationPush(ctx context.Context, intent work.Intent) error {292 var payload pushPayload293 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 err299 }300 type device struct {301 endpoint string302 subscription webpush.Subscription303 failures int304 }305 category := payload.Kind306 if !containsString(pushCategories, category) {307 category = pushRelay308 }309 var devices []device310 for rows.Next() {311 var item device312 var raw, categories string313 if err := rows.Scan(&item.endpoint, &raw, &item.failures, &categories); err != nil {314 rows.Close()315 return err316 }317 // A device that chose categories only wakes for those.318 if allowed := decodeCategories(categories); allowed != nil && !allowed[category] {319 continue320 }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 nil328 }329 keys, err := t.pushVAPIDKeys(ctx)330 if err != nil {331 return err332 }333 message, err := json.Marshal(t.pushMessage(ctx, payload))334 if err != nil {335 return err336 }337 client := t.pushClient338 if client == nil {339 client = replication.NewPinnedClient(10 * time.Second)340 }341 var result error342 delivered, removed, failed := 0, 0, 0343 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 result361}362363// pushMessage turns a notification summary into what the device shows.364func (t *Tenant) pushMessage(ctx context.Context, payload pushPayload) pushMessage {365 title := t.Policy().Name366 if title == "" {367 title = t.meta.Name368 }369 body := payload.Text370 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.Subject382 }383 url := payload.URL384 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}