| 1 | package daemon |
| 2 | |
| 3 | // Device notification triggers. Each stored event that addresses a member |
| 4 | // with registered devices becomes a short summary in one of five categories, |
| 5 | // which the person picks per device: private messages, replies to their own |
| 6 | // posts and repositories, mentions, requests for a decision, and relay |
| 7 | // notices. |
| 8 | |
| 9 | import ( |
| 10 | "context" |
| 11 | "encoding/json" |
| 12 | "strconv" |
| 13 | "strings" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 17 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 18 | ) |
| 19 | |
| 20 | const ( |
| 21 | pushMessages = "messages" |
| 22 | pushReplies = "replies" |
| 23 | pushMentions = "mentions" |
| 24 | pushApprovals = "approvals" |
| 25 | pushRelay = "relay" |
| 26 | pushCoalesce = 30 * time.Second |
| 27 | pushExcerptMax = 120 |
| 28 | ) |
| 29 | |
| 30 | var pushCategories = []string{pushMessages, pushReplies, pushMentions, pushApprovals, pushRelay} |
| 31 | |
| 32 | type pushNotice struct { |
| 33 | recipient, category, body, url string |
| 34 | // actions are the answers a device can give from the notification. |
| 35 | actions []pushAction |
| 36 | } |
| 37 | |
| 38 | // notifyDevices runs after an event is stored. It never fails the publish: |
| 39 | // a device summary is best effort on top of the durable event. |
| 40 | func (t *Tenant) notifyDevices(ctx context.Context, e event.Event) { |
| 41 | if !t.pushDevicesExist(ctx) { |
| 42 | return |
| 43 | } |
| 44 | for _, notice := range t.pushNotices(ctx, e) { |
| 45 | if !t.pushCoalesced(notice.recipient, notice.category) { |
| 46 | continue |
| 47 | } |
| 48 | if err := t.enqueuePushNotice(ctx, notice); err != nil { |
| 49 | t.app.telemetry.Logger().Debug("device notification not queued" , "error" , err) |
| 50 | } |
| 51 | } |
| 52 | } |
| 53 | |
| 54 | func (t *Tenant) pushDevicesExist(ctx context.Context) bool { |
| 55 | var count int |
| 56 | return t.store.DB().QueryRowContext(ctx, "SELECT count(*) FROM web_push" ).Scan(&count) == nil && count > 0 |
| 57 | } |
| 58 | |
| 59 | // pushCoalesced allows one summary per recipient and category within a short |
| 60 | // window, so a burst of replies produces a single notification. |
| 61 | func (t *Tenant) pushCoalesced(recipient, category string) bool { |
| 62 | t.pushMu.Lock() |
| 63 | defer t.pushMu.Unlock() |
| 64 | if t.pushRecent == nil { |
| 65 | t.pushRecent = make(map[string]time.Time) |
| 66 | } |
| 67 | key := recipient + "\x00" + category |
| 68 | now := time.Now() |
| 69 | if last, ok := t.pushRecent[key]; ok && now.Sub(last) < pushCoalesce { |
| 70 | return false |
| 71 | } |
| 72 | for k, at := range t.pushRecent { |
| 73 | if now.Sub(at) > pushCoalesce { |
| 74 | delete(t.pushRecent, k) |
| 75 | } |
| 76 | } |
| 77 | t.pushRecent[key] = now |
| 78 | return true |
| 79 | } |
| 80 | |
| 81 | func (t *Tenant) enqueuePushNotice(ctx context.Context, notice pushNotice) error { |
| 82 | return t.enqueuePushPayload(ctx, pushPayload{Recipient: notice.recipient, Kind: notice.category, Text: notice.body, URL: notice.url, Actions: notice.actions}) |
| 83 | } |
| 84 | |
| 85 | // pushNotices decides who an event should wake and why. |
| 86 | func (t *Tenant) pushNotices(ctx context.Context, e event.Event) []pushNotice { |
| 87 | base := strings.TrimRight(t.publicURL, "/" ) |
| 88 | var notices []pushNotice |
| 89 | // A request for a decision wakes the people asked in its own category |
| 90 | // and nowhere else, so one event yields one notification per device. |
| 91 | if kind, ok := approvalRequest(e); ok { |
| 92 | return t.approvalNotices(e, kind) |
| 93 | } |
| 94 | // A proposal from an agent asks the repository's maintainers for a |
| 95 | // decision the same way, and wakes nobody else. |
| 96 | if notices, ok := t.collaborationProposalNotices(ctx, e); ok { |
| 97 | return notices |
| 98 | } |
| 99 | // A job result, or feedback that needs the requester's attention, |
| 100 | // reaches the requester as a mention. |
| 101 | if status, ok := jobNotice(e); ok { |
| 102 | body := "job " + t.jobRequestKind(ctx, e) + " " + status |
| 103 | for _, recipient := range pushRecipients(e) { |
| 104 | notices = append(notices, pushNotice{recipient: recipient, category: pushMentions, body: body, url: base + "/e/" + e.ID}) |
| 105 | } |
| 106 | return notices |
| 107 | } |
| 108 | switch e.Kind { |
| 109 | case event.KIND_WRAP, 4: |
| 110 | for _, recipient := range pushRecipients(e) { |
| 111 | notices = append(notices, pushNotice{recipient: recipient, category: pushMessages, body: "New private message." , url: base + "/inbox" }) |
| 112 | } |
| 113 | case event.KIND_GIT_ISSUE, event.KIND_GIT_PR: |
| 114 | owner, identifier, ok := repoCoordinate(event.Tag(e, "a" )) |
| 115 | if !ok || owner == e.PubKey { |
| 116 | return nil |
| 117 | } |
| 118 | label := "New issue" |
| 119 | view := "issue" |
| 120 | if e.Kind == event.KIND_GIT_PR { |
| 121 | label, view = "New pull request" , "pr" |
| 122 | } |
| 123 | body := label |
| 124 | if subject := strings.TrimSpace(event.Tag(e, "subject" )); subject != "" { |
| 125 | body += ": " + excerpt(subject) |
| 126 | } |
| 127 | notices = append(notices, pushNotice{recipient: owner, category: pushReplies, body: body, url: base + "/repo?owner=" + owner + "&repo=" + identifier + "&view=" + view + "&id=" + e.ID}) |
| 128 | case 1, 1111, 30023: |
| 129 | authored := t.referencedAuthors(ctx, e) |
| 130 | for _, recipient := range pushRecipients(e) { |
| 131 | category, body := pushMentions, "Mentioned you: " |
| 132 | if authored[recipient] { |
| 133 | category, body = pushReplies, "Replied to you: " |
| 134 | } |
| 135 | notices = append(notices, pushNotice{recipient: recipient, category: category, body: body + excerpt(e.Content), url: base + "/e/" + e.ID}) |
| 136 | } |
| 137 | case event.KIND_CHAT, event.KIND_THREAD, event.KIND_THREAD_REPLY: |
| 138 | // Room messages name the room so the summary reads as a conversation. |
| 139 | roomID := event.Tag(e, "h" ) |
| 140 | if roomID == "" || e.Kind == event.KIND_MARMOT_GROUP { |
| 141 | return nil |
| 142 | } |
| 143 | room, err := t.community.Room(ctx, roomID) |
| 144 | if err != nil || !room.Live() { |
| 145 | return nil |
| 146 | } |
| 147 | name := room.Name |
| 148 | if name == "" { |
| 149 | name = room.ID |
| 150 | } |
| 151 | authored := t.referencedAuthors(ctx, e) |
| 152 | for _, recipient := range pushRecipients(e) { |
| 153 | category, body := pushMentions, "Mentioned you in " +name+": " |
| 154 | if authored[recipient] { |
| 155 | category, body = pushReplies, "Replied to you in " +name+": " |
| 156 | } |
| 157 | notices = append(notices, pushNotice{recipient: recipient, category: category, body: body + excerpt(e.Content), url: base + "/rooms/" + room.ID}) |
| 158 | } |
| 159 | } |
| 160 | return notices |
| 161 | } |
| 162 | |
| 163 | // jobNotice reports whether a long task event wakes its requester and with |
| 164 | // which status: every result does, and feedback does when it asks for |
| 165 | // payment or reports an error. Progress and success feedback stay quiet, |
| 166 | // since the result follows. |
| 167 | func jobNotice(e event.Event) (string, bool) { |
| 168 | if event.IsJobResult(e.Kind) { |
| 169 | return "result" , true |
| 170 | } |
| 171 | if e.Kind != event.KIND_JOB_FEEDBACK { |
| 172 | return "" , false |
| 173 | } |
| 174 | status := event.Tag(e, "status" ) |
| 175 | return status, status == "error" || status == "payment-required" |
| 176 | } |
| 177 | |
| 178 | // jobRequestKind names the request kind an answer belongs to: a result |
| 179 | // says it by its own kind, and feedback by the request the relay holds. |
| 180 | func (t *Tenant) jobRequestKind(ctx context.Context, e event.Event) string { |
| 181 | if event.IsJobResult(e.Kind) { |
| 182 | return strconv.Itoa(e.Kind - 1000) |
| 183 | } |
| 184 | id := event.Tag(e, "e" ) |
| 185 | if len(id) != 64 { |
| 186 | return strconv.Itoa(e.Kind) |
| 187 | } |
| 188 | rows, err := t.store.Query(ctx, event.Filter{IDs: []string{id}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 1}) |
| 189 | if err != nil || len(rows.Events) == 0 || !event.IsJobRequest(rows.Events[0].Kind) { |
| 190 | return strconv.Itoa(e.Kind) |
| 191 | } |
| 192 | return strconv.Itoa(rows.Events[0].Kind) |
| 193 | } |
| 194 | |
| 195 | // pushRecipients lists the addressed keys other than the author, bounded so |
| 196 | // a broadcast to a large list cannot fan out to every device. |
| 197 | func pushRecipients(e event.Event) []string { |
| 198 | seen := map[string]struct{}{e.PubKey: {}} |
| 199 | var recipients []string |
| 200 | for _, tag := range e.Tags { |
| 201 | if len(tag) < 2 || (tag[0] != "p" && tag[0] != "P" ) || len(tag[1]) != 64 { |
| 202 | continue |
| 203 | } |
| 204 | if _, ok := seen[tag[1]]; ok { |
| 205 | continue |
| 206 | } |
| 207 | seen[tag[1]] = struct{}{} |
| 208 | recipients = append(recipients, tag[1]) |
| 209 | if len(recipients) == 16 { |
| 210 | break |
| 211 | } |
| 212 | } |
| 213 | return recipients |
| 214 | } |
| 215 | |
| 216 | // referencedAuthors resolves the authors of events this one replies to, so a |
| 217 | // reply to your note counts as a reply rather than a mention. |
| 218 | func (t *Tenant) referencedAuthors(ctx context.Context, e event.Event) map[string]bool { |
| 219 | authors := map[string]bool{} |
| 220 | var ids []string |
| 221 | for _, name := range []string{"e" , "E" } { |
| 222 | for _, id := range event.TagValues(e, name) { |
| 223 | if len(id) == 64 && len(ids) < 8 { |
| 224 | ids = append(ids, id) |
| 225 | } |
| 226 | } |
| 227 | } |
| 228 | if len(ids) == 0 { |
| 229 | return authors |
| 230 | } |
| 231 | rows, err := t.store.Query(ctx, event.Filter{IDs: ids, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: len(ids)}) |
| 232 | if err != nil { |
| 233 | return authors |
| 234 | } |
| 235 | for _, row := range rows.Events { |
| 236 | authors[row.PubKey] = true |
| 237 | } |
| 238 | return authors |
| 239 | } |
| 240 | |
| 241 | func repoCoordinate(value string) (owner, identifier string, ok bool) { |
| 242 | parts := strings.SplitN(value, ":" , 3) |
| 243 | if len(parts) != 3 || parts[0] != strconv.Itoa(30617) || len(parts[1]) != 64 || parts[2] == "" { |
| 244 | return "" , "" , false |
| 245 | } |
| 246 | return parts[1], parts[2], true |
| 247 | } |
| 248 | |
| 249 | func excerpt(text string) string { |
| 250 | text = strings.Join(strings.Fields(text), " " ) |
| 251 | if len(text) > pushExcerptMax { |
| 252 | cut := pushExcerptMax |
| 253 | for cut > 0 && cut < len(text) && text[cut]&0xC0 == 0x80 { |
| 254 | cut-- |
| 255 | } |
| 256 | text = text[:cut] + "…" |
| 257 | } |
| 258 | return text |
| 259 | } |
| 260 | |
| 261 | // inboxSeen records when the person last opened their inbox, which bounds |
| 262 | // the unread count carried on the app badge. |
| 263 | func (t *Tenant) inboxSeen(ctx context.Context, pubkey string) int64 { |
| 264 | var at int64 |
| 265 | _ = t.store.GetSetting(ctx, "inbox.seen." +pubkey, &at) |
| 266 | return at |
| 267 | } |
| 268 | |
| 269 | func (t *Tenant) markInboxSeen(ctx context.Context, pubkey string) error { |
| 270 | return t.store.PutSetting(ctx, "inbox.seen." +pubkey, time.Now().Unix()) |
| 271 | } |
| 272 | |
| 273 | // inboxUnread counts events addressed to the key since it last opened the |
| 274 | // inbox, plus every request that still waits for the key's decision. It |
| 275 | // feeds the app badge; the page clears the conversation part on the next |
| 276 | // visit, and an answer clears the request. |
| 277 | func (t *Tenant) inboxUnread(ctx context.Context, pubkey string) int { |
| 278 | since := t.inboxSeen(ctx, pubkey) |
| 279 | // Count conversation kinds only, so relay records addressed to the |
| 280 | // owner do not inflate the badge. |
| 281 | filter := event.Filter{Kinds: []int{1, 4, event.KIND_CHAT, event.KIND_THREAD, event.KIND_THREAD_REPLY, 1111, event.KIND_WRAP, event.KIND_GIT_ISSUE, event.KIND_GIT_PR, 30023}, Tags: map[string][]string{"p" : {pubkey}}} |
| 282 | if since > 0 { |
| 283 | filter.Since = &since |
| 284 | } |
| 285 | count, err := t.store.Count(ctx, []event.Filter{filter}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{PubKeys: []string{pubkey}}}) |
| 286 | if err != nil { |
| 287 | return 0 |
| 288 | } |
| 289 | // An open request counts once: those newer than the last visit are |
| 290 | // already in the conversation count. |
| 291 | for _, item := range t.openApprovals(ctx, pubkey) { |
| 292 | if item.CreatedAt < since || !containsInt(filter.Kinds, item.Kind) { |
| 293 | count++ |
| 294 | } |
| 295 | } |
| 296 | if count > 99 { |
| 297 | return 99 |
| 298 | } |
| 299 | return int(count) |
| 300 | } |
| 301 | |
| 302 | func decodeCategories(raw string) map[string]bool { |
| 303 | if raw == "" { |
| 304 | return nil |
| 305 | } |
| 306 | var list []string |
| 307 | if json.Unmarshal([]byte(raw), &list) != nil { |
| 308 | return nil |
| 309 | } |
| 310 | allowed := map[string]bool{} |
| 311 | for _, item := range list { |
| 312 | allowed[item] = true |
| 313 | } |
| 314 | return allowed |
| 315 | } |