| 1 | package main |
| 2 | |
| 3 | // Mini-app server surface (authoring docs: web/static/apps/README.md). |
| 4 | // |
| 5 | // Two primitives, both E2E-preserving — the server never sees inside a blob: |
| 6 | // |
| 7 | // Scheduling POST/GET /api/schedule, DELETE /api/schedule/{id} |
| 8 | // A "future message": client-encrypted ciphertext held until deliver_at, |
| 9 | // then moved into the thread's message log and fanned out (WS + push) |
| 10 | // exactly like a posted message. Offline clients catch up from the log; |
| 11 | // the service worker decrypts the push and can show an app-authored |
| 12 | // notification. The server learns only (thread, time). |
| 13 | // |
| 14 | // State GET/PUT /api/threads/{id}/state |
| 15 | // One encrypted snapshot blob per (thread, slot) with optimistic |
| 16 | // versioning. Lets an app's folded state survive beyond the message |
| 17 | // window (the >100-events problem) and hold things like credentials — |
| 18 | // all ciphertext under the thread key, same trust boundary as messages. |
| 19 | |
| 20 | import ( |
| 21 | "encoding/json" |
| 22 | "log" |
| 23 | "net/http" |
| 24 | "time" |
| 25 | ) |
| 26 | |
| 27 | // maxStateBytes caps a state snapshot blob. Larger than a message (a snapshot |
| 28 | // aggregates many events) but still bounded so a misbehaving client can't grow |
| 29 | // the DB without limit. |
| 30 | const maxStateBytes = 256 * 1024 |
| 31 | |
| 32 | // maxPendingSchedules bounds the queue per thread — far above real use, low |
| 33 | // enough that a runaway client can't flood the table. |
| 34 | const maxPendingSchedules = 1000 |
| 35 | |
| 36 | // scheduleHorizon is how far ahead a delivery may be set. Two years is beyond |
| 37 | // any plausible reminder; anything longer is almost certainly a client bug. |
| 38 | const scheduleHorizon = 2 * 365 * 24 * time.Hour |
| 39 | |
| 40 | // memberOfThread resolves the caller's user id and checks thread membership — |
| 41 | // the standard gate on every mini-app endpoint (mirrors getMessages). |
| 42 | func (s *Server) memberOfThread(token, threadID string) (string, bool) { |
| 43 | if threadID == "" || !s.tokenOK(token) { |
| 44 | return "", false |
| 45 | } |
| 46 | userID, _ := s.db.deviceTokenUser(token) |
| 47 | if userID == "" || !s.db.threadMember(threadID, userID) { |
| 48 | return "", false |
| 49 | } |
| 50 | return userID, true |
| 51 | } |
| 52 | |
| 53 | // postSchedule queues a future message: {token, thread_id, deliver_at, blob}. |
| 54 | // The blob is already encrypted by the caller (the server can't check, only |
| 55 | // bound it). Returns {id} for later cancellation. |
| 56 | func (s *Server) postSchedule(w http.ResponseWriter, r *http.Request) { |
| 57 | var in struct { |
| 58 | Token string `json:"token"` |
| 59 | ThreadID string `json:"thread_id"` |
| 60 | DeliverAt int64 `json:"deliver_at"` |
| 61 | Blob string `json:"blob"` |
| 62 | } |
| 63 | r.Body = http.MaxBytesReader(w, r.Body, maxBlobBytes+4096) |
| 64 | if json.NewDecoder(r.Body).Decode(&in) != nil { |
| 65 | httpError(w, http.StatusBadRequest, "bad json") |
| 66 | return |
| 67 | } |
| 68 | userID, ok := s.memberOfThread(in.Token, in.ThreadID) |
| 69 | if !ok { |
| 70 | httpError(w, http.StatusForbidden, "not a thread member") |
| 71 | return |
| 72 | } |
| 73 | if in.Blob == "" || len(in.Blob) > maxBlobBytes { |
| 74 | httpError(w, http.StatusBadRequest, "bad blob") |
| 75 | return |
| 76 | } |
| 77 | now := time.Now() |
| 78 | // A small grace behind now absorbs clock skew; the sweep delivers it at once. |
| 79 | if in.DeliverAt < now.Add(-time.Minute).Unix() || in.DeliverAt > now.Add(scheduleHorizon).Unix() { |
| 80 | httpError(w, http.StatusBadRequest, "deliver_at out of range") |
| 81 | return |
| 82 | } |
| 83 | if s.db.countScheduledMessages(in.ThreadID) >= maxPendingSchedules { |
| 84 | httpError(w, http.StatusTooManyRequests, "too many pending schedules") |
| 85 | return |
| 86 | } |
| 87 | id, err := s.db.createScheduledMessage(in.ThreadID, in.Blob, userID, in.DeliverAt) |
| 88 | if err != nil { |
| 89 | httpError(w, http.StatusInternalServerError, "db error") |
| 90 | return |
| 91 | } |
| 92 | writeJSON(w, http.StatusOK, map[string]any{"id": id, "deliver_at": in.DeliverAt}) |
| 93 | } |
| 94 | |
| 95 | // getSchedules lists a thread's pending schedules ({id, deliver_at, blob}) so a |
| 96 | // client can reconcile app items with their schedule ids (e.g. a fresh device). |
| 97 | func (s *Server) getSchedules(w http.ResponseWriter, r *http.Request) { |
| 98 | q := r.URL.Query() |
| 99 | if _, ok := s.memberOfThread(q.Get("token"), q.Get("thread")); !ok { |
| 100 | httpError(w, http.StatusForbidden, "not a thread member") |
| 101 | return |
| 102 | } |
| 103 | list, err := s.db.listScheduledMessages(q.Get("thread")) |
| 104 | if err != nil { |
| 105 | httpError(w, http.StatusInternalServerError, "db error") |
| 106 | return |
| 107 | } |
| 108 | writeJSON(w, http.StatusOK, map[string]any{"schedules": list}) |
| 109 | } |
| 110 | |
| 111 | // deleteSchedule cancels a pending schedule. Any member of the schedule's thread |
| 112 | // may cancel (thread members share the app's state, so they share its timers). |
| 113 | // Cancelling an id that already fired (or never existed) succeeds quietly — |
| 114 | // cancellation races delivery by design, and "already gone" is not an error. |
| 115 | func (s *Server) deleteSchedule(w http.ResponseWriter, r *http.Request) { |
| 116 | token := r.URL.Query().Get("token") |
| 117 | id := r.PathValue("id") |
| 118 | tid, found, err := s.db.scheduledMessageThread(id) |
| 119 | if err != nil { |
| 120 | // A real DB failure must not masquerade as "already cancelled": a false |
| 121 | // 200 would let the client drop the reminder while it still fires. |
| 122 | httpError(w, http.StatusInternalServerError, "db error") |
| 123 | return |
| 124 | } |
| 125 | if found { |
| 126 | if _, ok := s.memberOfThread(token, tid); !ok { |
| 127 | httpError(w, http.StatusForbidden, "not a thread member") |
| 128 | return |
| 129 | } |
| 130 | if err := s.db.deleteScheduledMessage(id); err != nil { |
| 131 | httpError(w, http.StatusInternalServerError, "db error") |
| 132 | return |
| 133 | } |
| 134 | } else if !s.tokenOK(token) { |
| 135 | // Unknown id still requires a valid token — don't leak queue existence |
| 136 | // to strangers. |
| 137 | httpError(w, http.StatusForbidden, "invalid invite") |
| 138 | return |
| 139 | } |
| 140 | writeJSON(w, http.StatusOK, map[string]any{"ok": true}) |
| 141 | } |
| 142 | |
| 143 | // getNextDue reports the instance's earliest pending scheduled delivery: |
| 144 | // {due: <unix seconds>}, 0 when nothing is scheduled. It exists for the |
| 145 | // router's activator: at sleep time it records this so a hibernating instance |
| 146 | // is woken when the delivery comes due (activator.go's due-time index). |
| 147 | // Admin-gated — the payload is a single timestamp the server already stores in |
| 148 | // the clear (the schedule's (thread, time) metadata), but every data endpoint |
| 149 | // carries auth, and the activator already holds the instance's admin token via |
| 150 | // the env file it spawns instances with. |
| 151 | // |
| 152 | // The answer folds in the sidecar-reported charter due-time (memberbots.go |
| 153 | // postCharterNextDue), so member-bot charters — whose schedules live in |
| 154 | // blobs only the sidecar can read — wake a sleeping instance too. |
| 155 | func (s *Server) getNextDue(w http.ResponseWriter, r *http.Request) { |
| 156 | if !s.requireAdmin(w, r) { |
| 157 | return |
| 158 | } |
| 159 | due, err := s.db.nextScheduledDeliverAt() |
| 160 | if err != nil { |
| 161 | httpError(w, http.StatusInternalServerError, "db error") |
| 162 | return |
| 163 | } |
| 164 | writeJSON(w, http.StatusOK, map[string]any{ |
| 165 | "due": combineNextDue(due, s.db.charterNextDue(), time.Now().Unix()), |
| 166 | }) |
| 167 | } |
| 168 | |
| 169 | // combineNextDue folds the charter host's reported due-time into the schedule |
| 170 | // queue's earliest deliver_at: the sooner of the two wins, with 0 meaning |
| 171 | // "nothing" on either side. |
| 172 | // |
| 173 | // The asymmetry is deliberate: a PAST schedule deliver_at is honored (delivery |
| 174 | // consumes the row, so a past value only exists until the instance's next |
| 175 | // scheduler sweep — waking for it is exactly right), but a PAST charter |
| 176 | // due-time is ignored. The charter value has no consumption mechanism on the |
| 177 | // server — it changes only when the sidecar re-reports — and this probe runs |
| 178 | // at sleep time, after a full awake window. A value still in the past here |
| 179 | // means the sidecar had that whole window to act on it and re-report, and |
| 180 | // didn't (dead, keyless, or gone): honoring it would re-wake the instance |
| 181 | // immediately after every sleep, forever. Ignoring it caps the damage at the |
| 182 | // one wake the report legitimately bought and degrades to pre-index behavior |
| 183 | // (sleep until a request) — which is also CORRECT, not just safe: charters run |
| 184 | // in the sidecar, so waking an instance whose sidecar can't come up buys their |
| 185 | // schedules nothing. No clearing is needed — an ignored past value is inert, |
| 186 | // and any live sidecar overwrites it within one reconcile. (Same no-ping-pong |
| 187 | // reasoning as the activator's due-time index: wake/sleep cycles can never be |
| 188 | // spaced closer than the idle window, and a stale value never re-fires.) |
| 189 | func combineNextDue(schedDue, charterDue, now int64) int64 { |
| 190 | if charterDue <= 0 || charterDue < now { |
| 191 | return schedDue |
| 192 | } |
| 193 | if schedDue == 0 || charterDue < schedDue { |
| 194 | return charterDue |
| 195 | } |
| 196 | return schedDue |
| 197 | } |
| 198 | |
| 199 | // getThreadStateHandler returns a slot's snapshot: {blob, version}. version 0 + |
| 200 | // empty blob means nothing saved yet (the client starts a fresh fold). |
| 201 | func (s *Server) getThreadStateHandler(w http.ResponseWriter, r *http.Request) { |
| 202 | q := r.URL.Query() |
| 203 | threadID := r.PathValue("id") |
| 204 | if _, ok := s.memberOfThread(q.Get("token"), threadID); !ok { |
| 205 | httpError(w, http.StatusForbidden, "not a thread member") |
| 206 | return |
| 207 | } |
| 208 | slot := q.Get("slot") |
| 209 | if slot == "" { |
| 210 | httpError(w, http.StatusBadRequest, "missing slot") |
| 211 | return |
| 212 | } |
| 213 | blob, version, _, err := s.db.getThreadState(threadID, slot) |
| 214 | if err != nil { |
| 215 | httpError(w, http.StatusInternalServerError, "db error") |
| 216 | return |
| 217 | } |
| 218 | writeJSON(w, http.StatusOK, map[string]any{"blob": blob, "version": version}) |
| 219 | } |
| 220 | |
| 221 | // putThreadStateHandler writes a slot's snapshot with optimistic concurrency: |
| 222 | // {token, slot, blob, version} where version is what the caller last read (0 to |
| 223 | // create). A stale version gets 409 — reload, refold, retry (folds converge, so |
| 224 | // the retry is always safe). |
| 225 | func (s *Server) putThreadStateHandler(w http.ResponseWriter, r *http.Request) { |
| 226 | var in struct { |
| 227 | Token string `json:"token"` |
| 228 | Slot string `json:"slot"` |
| 229 | Blob string `json:"blob"` |
| 230 | Version int64 `json:"version"` |
| 231 | } |
| 232 | r.Body = http.MaxBytesReader(w, r.Body, maxStateBytes+4096) |
| 233 | if json.NewDecoder(r.Body).Decode(&in) != nil { |
| 234 | httpError(w, http.StatusBadRequest, "bad json") |
| 235 | return |
| 236 | } |
| 237 | threadID := r.PathValue("id") |
| 238 | if _, ok := s.memberOfThread(in.Token, threadID); !ok { |
| 239 | httpError(w, http.StatusForbidden, "not a thread member") |
| 240 | return |
| 241 | } |
| 242 | if in.Slot == "" || in.Blob == "" || len(in.Blob) > maxStateBytes { |
| 243 | httpError(w, http.StatusBadRequest, "bad slot or blob") |
| 244 | return |
| 245 | } |
| 246 | version, ok, err := s.db.putThreadState(threadID, in.Slot, in.Blob, in.Version) |
| 247 | if err != nil { |
| 248 | httpError(w, http.StatusInternalServerError, "db error") |
| 249 | return |
| 250 | } |
| 251 | if !ok { |
| 252 | httpError(w, http.StatusConflict, "version conflict") |
| 253 | return |
| 254 | } |
| 255 | writeJSON(w, http.StatusOK, map[string]any{"version": version}) |
| 256 | } |
| 257 | |
| 258 | // ---- delivery ------------------------------------------------------------ |
| 259 | |
| 260 | // runScheduler sweeps due schedules forever. Coarse (a few seconds) is plenty: |
| 261 | // reminders are minute-granularity, and each idle sweep is one indexed SELECT. |
| 262 | func (s *Server) runScheduler() { |
| 263 | for range time.Tick(5 * time.Second) { |
| 264 | s.sweepDueScheduled() |
| 265 | } |
| 266 | } |
| 267 | |
| 268 | // sweepDueScheduled delivers every due schedule: move it into the message log |
| 269 | // (transactional — see deliverScheduledMessage), then broadcast the standard |
| 270 | // "msg" frame and push. A delivered schedule is indistinguishable from a posted |
| 271 | // message, which is exactly what the client's fold model expects. Returns how |
| 272 | // many were delivered (used by tests, which call this directly — no ticker). |
| 273 | func (s *Server) sweepDueScheduled() int { |
| 274 | due, err := s.db.dueScheduledMessages(time.Now().Unix(), 100) |
| 275 | if err != nil { |
| 276 | log.Println("schedule: sweep:", err) |
| 277 | return 0 |
| 278 | } |
| 279 | n := 0 |
| 280 | for _, sm := range due { |
| 281 | // A DM closed since this was scheduled (issue #578): the conversation |
| 282 | // ended, so the queued message dies with it rather than forcing the |
| 283 | // thread back open. |
| 284 | if closed, _ := s.db.dmClosed(sm.ThreadID); closed { |
| 285 | _ = s.db.deleteScheduledMessage(sm.ID) |
| 286 | continue |
| 287 | } |
| 288 | m, ok, err := s.db.deliverScheduledMessage(sm) |
| 289 | if err != nil { |
| 290 | log.Println("schedule: deliver:", err) |
| 291 | continue |
| 292 | } |
| 293 | if !ok { |
| 294 | continue // raced another sweep; that one broadcasts |
| 295 | } |
| 296 | n++ |
| 297 | out, _ := json.Marshal(wsOut{Type: "msg", ThreadID: sm.ThreadID, ID: m.ID, CreatedAt: m.CreatedAt, Blob: m.Blob}) |
| 298 | s.broadcastToThread(sm.ThreadID, out) |
| 299 | go s.pushScheduled(m.Blob, sm.ThreadID, sm.AuthorID, m.ID, m.CreatedAt) |
| 300 | } |
| 301 | return n |
| 302 | } |
| 303 | |
| 304 | // pushScheduled is pushNewMessage for a scheduled delivery, with one deliberate |
| 305 | // difference: NO sender exclusion. The creator is exactly who wants this push (a |
| 306 | // reminder that skipped its own author would be useless). skipActive stays on — |
| 307 | // an open client receives the WS frame and the app's notify hook handles it. |
| 308 | func (s *Server) pushScheduled(blob, threadID, authorID string, id, at int64) { |
| 309 | if s.db.threadKind(threadID) == "space" { |
| 310 | return // the space thread never pushes — see pushNewMessage |
| 311 | } |
| 312 | members, err := s.db.memberUserIDs(threadID) |
| 313 | if err != nil { |
| 314 | log.Println("schedule: push members:", err) |
| 315 | return |
| 316 | } |
| 317 | set := make(map[string]bool, len(members)) |
| 318 | for _, u := range members { |
| 319 | set[u] = true |
| 320 | } |
| 321 | s.dropMutedRecipients(threadID, set) // mute wins even over a reminder |
| 322 | if authorID != "" { |
| 323 | s.dropBlockingRecipients(authorID, set) // a blocked author never buzzes the blocker |
| 324 | } |
| 325 | payload, _ := json.Marshal(map[string]string{"blob": blob, "thread": threadID}) |
| 326 | s.sendToRecipients(payload, "sched", "", true, set) |
| 327 | s.apnsFanOut(threadID, blob, id, at, "", true, set) |
| 328 | } |
| 329 | |
amadan