| 1 | package main |
| 2 | |
| 3 | // Presence: who's around right now. Effective state is computed server-side |
| 4 | // from three inputs — open sockets, the member's pinned "away" flag, and the |
| 5 | // existing per-user activity signal (the "active" frames that already drive |
| 6 | // push suppression, hub.go): |
| 7 | // |
| 8 | // offline — no open sockets |
| 9 | // away — connected, but pinned away by the member OR idle >10 min |
| 10 | // online — connected and recently active |
| 11 | // |
| 12 | // Privacy: presence is connection metadata the server already possesses (it |
| 13 | // holds the sockets and the activity timestamps); no content and no key rides |
| 14 | // on it. The pinned flag is a per-user enum like `role`. Fan-out is scoped to |
| 15 | // the subject's connections — the same people who can see them in the roster — |
| 16 | // so nobody learns the presence of a member they can't already see. |
| 17 | |
| 18 | import ( |
| 19 | "encoding/json" |
| 20 | "net/http" |
| 21 | "sync" |
| 22 | "time" |
| 23 | ) |
| 24 | |
| 25 | const ( |
| 26 | presenceOnline = "online" |
| 27 | presenceAway = "away" |
| 28 | presenceOffline = "offline" |
| 29 | ) |
| 30 | |
| 31 | // presenceIdleAfter: a connected person with no "active" signal for this long |
| 32 | // counts as away (auto-idle). A var so tests can shrink it. |
| 33 | var presenceIdleAfter = 10 * time.Minute |
| 34 | |
| 35 | // presenceTracker owns the socket counts and the last state broadcast per |
| 36 | // user, so only *transitions* emit frames. Sockets are counted here (not in |
| 37 | // the hub) by handleWS's connect/defer, which avoids racing the hub goroutine: |
| 38 | // however a socket dies — clean close, network error, or the hub dropping a |
| 39 | // backed-up client — the reader's defer runs and decrements. |
| 40 | type presenceTracker struct { |
| 41 | mu sync.Mutex |
| 42 | conns map[string]int // user id → open sockets |
| 43 | last map[string]string // user id → last broadcast state |
| 44 | } |
| 45 | |
| 46 | // addConn counts a new socket for uid; reports whether it's their first. |
| 47 | func (p *presenceTracker) addConn(uid string) { |
| 48 | if uid == "" { |
| 49 | return |
| 50 | } |
| 51 | p.mu.Lock() |
| 52 | defer p.mu.Unlock() |
| 53 | if p.conns == nil { |
| 54 | p.conns = map[string]int{} |
| 55 | } |
| 56 | p.conns[uid]++ |
| 57 | } |
| 58 | |
| 59 | func (p *presenceTracker) dropConn(uid string) { |
| 60 | if uid == "" { |
| 61 | return |
| 62 | } |
| 63 | p.mu.Lock() |
| 64 | defer p.mu.Unlock() |
| 65 | if p.conns[uid] > 1 { |
| 66 | p.conns[uid]-- |
| 67 | } else { |
| 68 | delete(p.conns, uid) |
| 69 | } |
| 70 | } |
| 71 | |
| 72 | func (p *presenceTracker) connected(uid string) bool { |
| 73 | p.mu.Lock() |
| 74 | defer p.mu.Unlock() |
| 75 | return p.conns[uid] > 0 |
| 76 | } |
| 77 | |
| 78 | // connectedUsers snapshots who holds at least one socket (for the idle sweep |
| 79 | // and for stamping a roster response in one pass). |
| 80 | func (p *presenceTracker) connectedUsers() map[string]bool { |
| 81 | p.mu.Lock() |
| 82 | defer p.mu.Unlock() |
| 83 | out := make(map[string]bool, len(p.conns)) |
| 84 | for uid := range p.conns { |
| 85 | out[uid] = true |
| 86 | } |
| 87 | return out |
| 88 | } |
| 89 | |
| 90 | // setLast records the state about to be broadcast; reports whether it changed. |
| 91 | func (p *presenceTracker) setLast(uid, state string) bool { |
| 92 | p.mu.Lock() |
| 93 | defer p.mu.Unlock() |
| 94 | if p.last == nil { |
| 95 | p.last = map[string]string{} |
| 96 | } |
| 97 | // A user nobody has seen transition yet starts implicitly offline. |
| 98 | prev := p.last[uid] |
| 99 | if prev == "" { |
| 100 | prev = presenceOffline |
| 101 | } |
| 102 | if prev == state { |
| 103 | return false |
| 104 | } |
| 105 | p.last[uid] = state |
| 106 | return true |
| 107 | } |
| 108 | |
| 109 | // presenceState resolves one user's effective state from its inputs. Agents |
| 110 | // never idle: a bot sidecar holds a socket but sends no "active" frames, and a |
| 111 | // connected bot is simply online. lastActive is unix seconds (0 = never). |
| 112 | func presenceState(connected bool, pinned string, isAgent bool, lastActive int64) string { |
| 113 | switch { |
| 114 | case !connected: |
| 115 | return presenceOffline |
| 116 | case pinned == presenceAway: |
| 117 | return presenceAway |
| 118 | case !isAgent && time.Since(time.Unix(lastActive, 0)) > presenceIdleAfter: |
| 119 | return presenceAway |
| 120 | default: |
| 121 | return presenceOnline |
| 122 | } |
| 123 | } |
| 124 | |
| 125 | func (s *Server) lastActive(uid string) int64 { |
| 126 | if v, ok := s.activeUsers.Load(uid); ok { |
| 127 | return v.(int64) |
| 128 | } |
| 129 | return 0 |
| 130 | } |
| 131 | |
| 132 | func (s *Server) effectivePresence(uid string) string { |
| 133 | if uid == "" { |
| 134 | return presenceOffline |
| 135 | } |
| 136 | pinned, isAgent := s.db.userPresenceInfo(uid) |
| 137 | return presenceState(s.presence.connected(uid), pinned, isAgent, s.lastActive(uid)) |
| 138 | } |
| 139 | |
| 140 | // presenceChanged recomputes uid's state and, if it differs from the last one |
| 141 | // broadcast, fans a presence frame out to their connections (self included — |
| 142 | // their other devices keep the toggle in sync). |
| 143 | func (s *Server) presenceChanged(uid string) { |
| 144 | if uid == "" { |
| 145 | return |
| 146 | } |
| 147 | state := s.effectivePresence(uid) |
| 148 | if !s.presence.setLast(uid, state) { |
| 149 | return |
| 150 | } |
| 151 | entries, err := s.db.connections(uid) |
| 152 | if err != nil { |
| 153 | return |
| 154 | } |
| 155 | set := make(map[string]bool, len(entries)) |
| 156 | for _, e := range entries { |
| 157 | set[e.UserID] = true |
| 158 | } |
| 159 | out, _ := json.Marshal(map[string]any{"type": "presence", "user_id": uid, "presence": state}) |
| 160 | s.hub.broadcast <- outbound{members: set, data: out} |
| 161 | } |
| 162 | |
| 163 | // runPresenceSweeper catches the transitions nothing else triggers — a |
| 164 | // connected member going idle (their "active" beats just stop). Tests call |
| 165 | // sweepPresence directly; only the real server runs the ticker. |
| 166 | func (s *Server) runPresenceSweeper() { |
| 167 | for range time.Tick(30 * time.Second) { |
| 168 | s.sweepPresence() |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | func (s *Server) sweepPresence() { |
| 173 | for uid := range s.presence.connectedUsers() { |
| 174 | s.presenceChanged(uid) |
| 175 | } |
| 176 | } |
| 177 | |
| 178 | // stampPresence overwrites each roster entry's presence (the raw pinned flag |
| 179 | // as scanned) with the caller-visible effective state, so the roster clients |
| 180 | // already load doubles as the presence snapshot. |
| 181 | func (s *Server) stampPresence(entries []RosterEntry) { |
| 182 | connected := s.presence.connectedUsers() |
| 183 | for i := range entries { |
| 184 | e := &entries[i] |
| 185 | e.Presence = presenceState(connected[e.UserID], e.Presence, e.IsAgent, s.lastActive(e.UserID)) |
| 186 | } |
| 187 | } |
| 188 | |
| 189 | // postPresence pins or unpins the caller's Away status. Stored per user (not |
| 190 | // per device), so it follows them across devices; the transition broadcasts |
| 191 | // like any other. |
| 192 | func (s *Server) postPresence(w http.ResponseWriter, r *http.Request) { |
| 193 | var body struct { |
| 194 | Token string `json:"token"` |
| 195 | Away bool `json:"away"` |
| 196 | } |
| 197 | if json.NewDecoder(r.Body).Decode(&body) != nil { |
| 198 | httpError(w, http.StatusBadRequest, "bad request") |
| 199 | return |
| 200 | } |
| 201 | if !s.tokenOK(body.Token) { |
| 202 | httpError(w, http.StatusForbidden, "not a member") |
| 203 | return |
| 204 | } |
| 205 | uid, ok := s.db.deviceTokenUser(body.Token) |
| 206 | if !ok { |
| 207 | httpError(w, http.StatusBadRequest, "bad request") |
| 208 | return |
| 209 | } |
| 210 | pinned := "" |
| 211 | if body.Away { |
| 212 | pinned = presenceAway |
| 213 | } |
| 214 | if err := s.db.setUserPresence(uid, pinned); err != nil { |
| 215 | httpError(w, http.StatusInternalServerError, "db error") |
| 216 | return |
| 217 | } |
| 218 | s.presenceChanged(uid) |
| 219 | writeJSON(w, http.StatusOK, map[string]any{"ok": true, "away": body.Away}) |
| 220 | } |
| 221 | |
amadan