| 1 | // live.go is the transport a browser edits a deck over: the update log |
| 2 | // by cursor, one new update in, an advisory stream out, and a heartbeat |
| 3 | // saying who is in the deck. |
| 4 | // |
| 5 | // It is Docs' live-sync design adopted whole (its internal/docs/{ |
| 6 | // handlers.go, stream.go}), with one addition and one difference. |
| 7 | // |
| 8 | // The addition is presence, which is NOT document state (spec §3.3): it |
| 9 | // lives in the hub's memory, never in the log, and is gone when this |
| 10 | // process is. |
| 11 | // |
| 12 | // The difference is framing. Docs concatenates updates, because a Yjs |
| 13 | // update is self-delimiting and its client feeds the whole blob to one |
| 14 | // decoder. Slides frames each one as a 4-byte big-endian length then |
| 15 | // the bytes, so the client can hand them to the core one at a time and |
| 16 | // so a fixture can compare a frame against the update that produced it |
| 17 | // (the plan's live-sync wire). Concatenation would make both of those |
| 18 | // "decode and hope". |
| 19 | // |
| 20 | // What never travels on the stream is bytes. The stream says "there is |
| 21 | // content through seq N" and the client fetches it through the endpoint |
| 22 | // it already used at mount — one code path from stored update to a |
| 23 | // browser's Y.Doc rather than two that can drift, and no base64 tax on |
| 24 | // every keystroke batch. |
| 25 | package slides |
| 26 | |
| 27 | import ( |
| 28 | "context" |
| 29 | "encoding/binary" |
| 30 | "encoding/json" |
| 31 | "errors" |
| 32 | "fmt" |
| 33 | "io" |
| 34 | "net/http" |
| 35 | "strconv" |
| 36 | "time" |
| 37 | |
| 38 | "gorm.io/gorm" |
| 39 | |
| 40 | "amadan.net/rastrillo/rastrillo" |
| 41 | |
| 42 | "amadan.net/oficina/slides/internal/slides/core" |
| 43 | "amadan.net/oficina/slides/internal/slides/deck" |
| 44 | "amadan.net/oficina/slides/internal/slides/live" |
| 45 | ) |
| 46 | |
| 47 | // maxUpdateBytes is the cap on one posted update (the plan's 1 MiB). |
| 48 | // Generous for a keystroke batch and for a pasted slide; small enough |
| 49 | // that a client cannot make this instance hold an arbitrary amount of |
| 50 | // memory by opening one connection. |
| 51 | const maxUpdateBytes = 1 << 20 |
| 52 | |
| 53 | // maxPresenceBytes is the cap on one heartbeat body. A heartbeat is |
| 54 | // four short strings; anything approaching this is not one. |
| 55 | const maxPresenceBytes = 4 << 10 |
| 56 | |
| 57 | // maxPresenceField bounds each string the client supplies. Ids in this |
| 58 | // app are twelve characters (decks.go's newID) and a client id is a |
| 59 | // browser's own random string, so 128 is roomy — and a cap of some kind |
| 60 | // is required, because these strings are held in memory for the life of |
| 61 | // a heartbeat by a request that only had to be signed in. |
| 62 | const maxPresenceField = 128 |
| 63 | |
| 64 | // errUpdateRefused is the core looking at the bytes a client posted and |
| 65 | // saying no: malformed wire bytes, or bytes that trapped the module and |
| 66 | // were contained. It is the client's problem, so it is a 422 and an |
| 67 | // Info-level log — accepting arbitrary input over HTTP and occasionally |
| 68 | // being sent nonsense is the ordinary case, not an incident. |
| 69 | // |
| 70 | // Everything else out of appendUpdate — a database failure, a replay of |
| 71 | // this deck's own history going wrong, a rebuild that itself failed — |
| 72 | // is a 500, because the client did not cause it and an operator needs |
| 73 | // to see it. |
| 74 | var errUpdateRefused = errors.New("slides: the core refused this update") |
| 75 | |
| 76 | // --- GET /d/{sqid}/updates --- |
| 77 | |
| 78 | // updatesGet serves the deck's log from a cursor: `application/octet- |
| 79 | // stream`, each update framed as a 4-byte big-endian length then the |
| 80 | // bytes, in seq order. |
| 81 | // |
| 82 | // `Slides-Seq` is the highest seq in the BODY, or the cursor itself |
| 83 | // when the body is empty — derived from the bytes in hand and never |
| 84 | // from a second MAX(seq) query. The difference matters: a separate |
| 85 | // query can see an update this response does not carry, and the client |
| 86 | // would then record a cursor for bytes it never received and never ask |
| 87 | // for them again. |
| 88 | // |
| 89 | // `Slides-Snapshot: <seq>` says the first frame is a snapshot rather |
| 90 | // than a log row. A client that does not care can ignore it — a |
| 91 | // snapshot is a Yjs update like any other and applying it is correct |
| 92 | // either way — and one that does can tell "here is the deck" from "here |
| 93 | // is what changed". |
| 94 | func (a *app) updatesGet(w http.ResponseWriter, r *http.Request) { |
| 95 | u, ok := a.viewer(w, r) |
| 96 | if !ok { |
| 97 | return |
| 98 | } |
| 99 | d, ok := a.deckFor(w, r, u) |
| 100 | if !ok { |
| 101 | return |
| 102 | } |
| 103 | since, ok := parseSince(w, r) |
| 104 | if !ok { |
| 105 | return |
| 106 | } |
| 107 | |
| 108 | updates, highest, snapshotSeq, err := a.readUpdates(r.Context(), d, since) |
| 109 | if err != nil { |
| 110 | a.fail(w, r, "read a deck's log", err) |
| 111 | return |
| 112 | } |
| 113 | |
| 114 | w.Header().Set("Content-Type", "application/octet-stream") |
| 115 | w.Header().Set("Cache-Control", "no-store") |
| 116 | w.Header().Set("Slides-Seq", strconv.FormatInt(highest, 10)) |
| 117 | if snapshotSeq > 0 { |
| 118 | w.Header().Set("Slides-Snapshot", strconv.FormatInt(snapshotSeq, 10)) |
| 119 | } |
| 120 | |
| 121 | var length [4]byte |
| 122 | for _, u := range updates { |
| 123 | binary.BigEndian.PutUint32(length[:], uint32(len(u))) |
| 124 | if _, err := w.Write(length[:]); err != nil { |
| 125 | // The client went away mid-response. Logged rather than |
| 126 | // surfaced: the status line is already on the wire. |
| 127 | a.logger.Error("slides: write the update log", "deck", d.Sqid, "err", err) |
| 128 | return |
| 129 | } |
| 130 | if _, err := w.Write(u); err != nil { |
| 131 | a.logger.Error("slides: write the update log", "deck", d.Sqid, "err", err) |
| 132 | return |
| 133 | } |
| 134 | } |
| 135 | } |
| 136 | |
| 137 | // readUpdates is what a cursor is owed: the frames to send, the highest |
| 138 | // seq they carry, and the seq of the snapshot when one of them is one. |
| 139 | // |
| 140 | // The snapshot is only ever offered at since=0. A client mid-session |
| 141 | // holds a prefix of the log and needs the tail after it; handing it a |
| 142 | // whole-document update instead would be correct (Yjs would merge it) |
| 143 | // and wasteful on every single fetch. At since=0 it is the opposite: a |
| 144 | // deck with two thousand updates behind it would otherwise send all two |
| 145 | // thousand to every tab that opens it. |
| 146 | func (a *app) readUpdates(ctx context.Context, d Deck, since int64) (frames [][]byte, highest, snapshotSeq int64, err error) { |
| 147 | highest = since |
| 148 | tail := since |
| 149 | |
| 150 | if since == 0 { |
| 151 | // An accelerator, never the source of truth (snapshot.go): a |
| 152 | // lookup that fails serves the plain log rather than failing a |
| 153 | // request the log alone can still answer. |
| 154 | snap, err := a.latestSnapshot(ctx, d.ID) |
| 155 | if err != nil { |
| 156 | a.logger.Error("slides: load a deck's newest snapshot, serving the plain log instead", |
| 157 | "deck", d.Sqid, "err", err) |
| 158 | } else if snap != nil && snap.Seq > since { |
| 159 | frames = append(frames, snap.Bytes) |
| 160 | snapshotSeq = snap.Seq |
| 161 | tail = snap.Seq |
| 162 | highest = snap.Seq |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | var updates []DeckUpdate |
| 167 | if err := a.db.G.WithContext(ctx). |
| 168 | Where("deck_id = ? AND seq > ?", d.ID, tail). |
| 169 | Order("seq").Find(&updates).Error; err != nil { |
| 170 | return nil, 0, 0, fmt.Errorf("slides: read deck %s log from %d: %w", d.Sqid, tail, err) |
| 171 | } |
| 172 | for _, u := range updates { |
| 173 | frames = append(frames, u.Bytes) |
| 174 | if u.Seq > highest { |
| 175 | highest = u.Seq |
| 176 | } |
| 177 | } |
| 178 | return frames, highest, snapshotSeq, nil |
| 179 | } |
| 180 | |
| 181 | // parseSince reads the ?since= cursor: the highest seq the client |
| 182 | // already holds, so the answer carries only what came after it. Absent |
| 183 | // means 0, which asks for the whole deck — what a tab wants at mount. |
| 184 | // |
| 185 | // A malformed or negative value is a 400 rather than a silent 0. |
| 186 | // Treating "banana" as "send everything" would answer an expensive |
| 187 | // question nobody asked and hide a client bug behind a response that |
| 188 | // looks fine. |
| 189 | func parseSince(w http.ResponseWriter, r *http.Request) (int64, bool) { |
| 190 | raw := r.URL.Query().Get("since") |
| 191 | if raw == "" { |
| 192 | return 0, true |
| 193 | } |
| 194 | since, err := strconv.ParseInt(raw, 10, 64) |
| 195 | if err != nil || since < 0 { |
| 196 | http.Error(w, rastrillo.T(r, "slides.live.error.badsince"), http.StatusBadRequest) |
| 197 | return 0, false |
| 198 | } |
| 199 | return since, true |
| 200 | } |
| 201 | |
| 202 | // --- POST /d/{sqid}/updates --- |
| 203 | |
| 204 | // updatesPost takes one Yjs update from a browser, applies it through |
| 205 | // the core, stores it and answers 204 with the seq it landed at. |
| 206 | // |
| 207 | // The body is wrapped in MaxBytesReader before anything reads it, so an |
| 208 | // oversized update is refused before it is buffered into memory at all, |
| 209 | // never after. |
| 210 | func (a *app) updatesPost(w http.ResponseWriter, r *http.Request) { |
| 211 | u, ok := a.viewer(w, r) |
| 212 | if !ok { |
| 213 | return |
| 214 | } |
| 215 | d, ok := a.deckFor(w, r, u) |
| 216 | if !ok { |
| 217 | return |
| 218 | } |
| 219 | |
| 220 | r.Body = http.MaxBytesReader(w, r.Body, maxUpdateBytes) |
| 221 | update, err := io.ReadAll(r.Body) |
| 222 | if err != nil { |
| 223 | var tooBig *http.MaxBytesError |
| 224 | if errors.As(err, &tooBig) { |
| 225 | http.Error(w, rastrillo.T(r, "slides.live.error.toolarge"), http.StatusRequestEntityTooLarge) |
| 226 | return |
| 227 | } |
| 228 | a.fail(w, r, "read a posted update", err) |
| 229 | return |
| 230 | } |
| 231 | |
| 232 | seq, err := a.appendUpdate(r.Context(), &d, actorFor(u), update) |
| 233 | if err != nil { |
| 234 | if errors.Is(err, errUpdateRefused) { |
| 235 | a.logger.Info("slides: refuse a posted update", "deck", d.Sqid, "err", err) |
| 236 | http.Error(w, rastrillo.T(r, "slides.live.error.invalid"), http.StatusUnprocessableEntity) |
| 237 | return |
| 238 | } |
| 239 | if errors.Is(err, errDeckBusy) { |
| 240 | // Two retries both lost: real contention rather than a |
| 241 | // race worth spinning on. The client's own queue reposts, |
| 242 | // and Yjs makes a duplicate harmless. |
| 243 | http.Error(w, rastrillo.T(r, "slides.live.error.busy"), http.StatusConflict) |
| 244 | return |
| 245 | } |
| 246 | a.fail(w, r, "apply a posted update", err) |
| 247 | return |
| 248 | } |
| 249 | |
| 250 | // The header before the broadcast, both only on success: an |
| 251 | // announced seq must be one a reader can actually fetch back. |
| 252 | w.Header().Set("Slides-Seq", strconv.FormatInt(seq, 10)) |
| 253 | w.WriteHeader(http.StatusNoContent) |
| 254 | } |
| 255 | |
| 256 | // appendUpdate is the second write path into a document, beside |
| 257 | // applyOps, and it takes the same lock the same way: open the deck, |
| 258 | // apply the update to the handle, insert the log row at Seq+1 and bump |
| 259 | // the row's seq in ONE transaction whose `WHERE id = ? AND seq = ?` is |
| 260 | // the optimistic lock. One retry on errDeckBusy from a freshly opened |
| 261 | // document, for editDeck's reason — the batch applied nothing, so |
| 262 | // replaying it against the state that beat us is correct. |
| 263 | // |
| 264 | // It exists rather than being folded into applyOps because the two |
| 265 | // differ in exactly one line and in what a failure of that line MEANS. |
| 266 | // applyOps hands the core a batch this app wrote, so a refusal is a bug |
| 267 | // here; appendUpdate hands it bytes off the wire, so a refusal is the |
| 268 | // ordinary case and has to come back distinguishable (errUpdateRefused) |
| 269 | // rather than as a 500. |
| 270 | // |
| 271 | // The one line is previewUpdate, and it does more than apply: it reads |
| 272 | // the deck back and parses it BEFORE anything is stored. Bytes off the |
| 273 | // wire can put any shape at any key, and a shape the typed reader |
| 274 | // cannot hold would otherwise be committed to the log and then fail |
| 275 | // every openDeck after it — the viewer, the editor, every slide, every |
| 276 | // asset, and both write paths, for good. The view itself tolerates |
| 277 | // the shapes it can (core/src/view.rs); this is the gate for whatever |
| 278 | // it cannot, and it stands whether or not the view is clever. |
| 279 | // |
| 280 | // d is a pointer because its Seq is updated in place on success and |
| 281 | // re-read on a conflict, exactly as editDeck does. |
| 282 | func (a *app) appendUpdate(ctx context.Context, d *Deck, actor string, update []byte) (int64, error) { |
| 283 | var lastErr error |
| 284 | for attempt := 0; attempt < 2; attempt++ { |
| 285 | h, _, err := a.openDeck(ctx, *d) |
| 286 | if err != nil { |
| 287 | return 0, err |
| 288 | } |
| 289 | view, err := a.previewUpdate(ctx, h, d, update) |
| 290 | if err != nil { |
| 291 | h.Close(ctx) |
| 292 | return 0, err |
| 293 | } |
| 294 | seq, err := a.commitUpdate(ctx, d, actor, update) |
| 295 | if err == nil { |
| 296 | // After the commit, in this order and for the same reasons |
| 297 | // applyOps has: nobody is told about a seq a rolled-back |
| 298 | // transaction never wrote, and the snapshot is taken from |
| 299 | // a handle that holds exactly what the row now claims. |
| 300 | a.live.PublishSeq(d.Sqid, seq) |
| 301 | a.mirrorMeta(ctx, d, view, seq) |
| 302 | a.maybeSnapshot(ctx, *d, h, seq) |
| 303 | h.Close(ctx) |
| 304 | return seq, nil |
| 305 | } |
| 306 | h.Close(ctx) |
| 307 | |
| 308 | lastErr = err |
| 309 | if !errors.Is(err, errDeckBusy) { |
| 310 | return 0, err |
| 311 | } |
| 312 | // Re-read the row: our copy's Seq is what just lost. |
| 313 | if err := a.db.G.WithContext(ctx).First(d, d.ID).Error; err != nil { |
| 314 | return 0, fmt.Errorf("slides: reload deck %s: %w", d.Sqid, err) |
| 315 | } |
| 316 | } |
| 317 | return 0, lastErr |
| 318 | } |
| 319 | |
| 320 | // previewUpdate applies the posted bytes to a freshly opened handle and |
| 321 | // reads the deck they produce, outside any transaction: the handle is |
| 322 | // per-attempt and discarded on any failure, so nothing here has to be |
| 323 | // undone. All three refusals are errUpdateRefused — the core would not |
| 324 | // take the bytes, or it took them and the text they leave is outside |
| 325 | // the schema, or the deck they make cannot be read — because all three |
| 326 | // are the client's problem and none has touched the log. Anything else |
| 327 | // is a 500 for the reasons the variable's own comment gives. |
| 328 | // |
| 329 | // The check comes BEFORE the parse and is the stronger of the two |
| 330 | // (codex 3). The view is a projection, and a tolerant one: it drops a |
| 331 | // formatting attribute it does not know and reads the words around it, |
| 332 | // so an update carrying `unknown_mark` on a run parses perfectly. The |
| 333 | // editor does not read the projection. It binds y-prosemirror to the |
| 334 | // shared text itself, and y-prosemirror 1.3.7 answers a mark its schema |
| 335 | // lacks by deleting the whole XmlText — the words, not the mark. So the |
| 336 | // gate has to look at the document, not at what the document projects |
| 337 | // to: core.Handle.Check walks every text fragment and refuses anything |
| 338 | // the §2.6 vocabulary does not define. The parse stays for what it was |
| 339 | // always for — a shape the typed reader cannot hold at all. |
| 340 | // |
| 341 | // The view comes back so mirrorMeta can use it rather than reading the |
| 342 | // deck out of the module a second time on every keystroke batch. |
| 343 | func (a *app) previewUpdate(ctx context.Context, h *core.Handle, d *Deck, update []byte) (*deck.Deck, error) { |
| 344 | if err := h.Apply(ctx, update); err != nil { |
| 345 | if core.Rejected(err) || core.Recovered(err) { |
| 346 | return nil, fmt.Errorf("%w: %w", errUpdateRefused, err) |
| 347 | } |
| 348 | // A rebuild that itself failed, or a full retained log: |
| 349 | // neither is the client's doing, and the second means this |
| 350 | // deck needs a snapshot before it can take anything else. |
| 351 | return nil, fmt.Errorf("slides: apply a posted update to deck %s: %w", d.Sqid, err) |
| 352 | } |
| 353 | if err := h.Check(ctx); err != nil { |
| 354 | if core.Rejected(err) || core.Recovered(err) { |
| 355 | return nil, fmt.Errorf("%w: the text it leaves is outside the schema: %w", errUpdateRefused, err) |
| 356 | } |
| 357 | return nil, fmt.Errorf("slides: check a posted update to deck %s: %w", d.Sqid, err) |
| 358 | } |
| 359 | raw, err := h.JSON(ctx) |
| 360 | if err != nil { |
| 361 | return nil, fmt.Errorf("slides: read deck %s after a posted update: %w", d.Sqid, err) |
| 362 | } |
| 363 | view, err := deck.Parse(raw) |
| 364 | if err != nil { |
| 365 | return nil, fmt.Errorf("%w: the deck it produces cannot be read: %w", errUpdateRefused, err) |
| 366 | } |
| 367 | return view, nil |
| 368 | } |
| 369 | |
| 370 | // commitUpdate is appendUpdate's one attempt at the database: applyOps' |
| 371 | // transaction with the update already applied and checked by |
| 372 | // previewUpdate, so the transaction holds SQLite's single writer for |
| 373 | // two statements and no guest calls. |
| 374 | func (a *app) commitUpdate(ctx context.Context, d *Deck, actor string, update []byte) (int64, error) { |
| 375 | next := d.Seq + 1 |
| 376 | err := a.db.G.WithContext(ctx).Transaction(func(tx *gorm.DB) error { |
| 377 | if err := tx.Create(&DeckUpdate{ |
| 378 | DeckID: d.ID, |
| 379 | Seq: next, |
| 380 | Actor: actor, |
| 381 | Bytes: update, |
| 382 | CreatedAt: time.Now().UTC(), |
| 383 | }).Error; err != nil { |
| 384 | if errors.Is(err, gorm.ErrDuplicatedKey) { |
| 385 | // Somebody else's update is already at this seq: the |
| 386 | // lost race, reported as the lost race. |
| 387 | return errDeckBusy |
| 388 | } |
| 389 | return fmt.Errorf("slides: record deck %s update %d: %w", d.Sqid, next, err) |
| 390 | } |
| 391 | |
| 392 | // Only the seq here. The row's title and theme follow the |
| 393 | // document too, but that is mirrorMeta's job AFTER the commit: |
| 394 | // it costs a read of the whole deck out of the wasm module, |
| 395 | // and metaRowUpdates' own comment says why that must not |
| 396 | // happen inside this transaction — it would hold SQLite's |
| 397 | // single writer for the length of a guest call on the path of |
| 398 | // every keystroke batch in the app. |
| 399 | res := tx.Model(&Deck{}).Where("id = ? AND seq = ?", d.ID, d.Seq). |
| 400 | Updates(map[string]any{"seq": next}) |
| 401 | if res.Error != nil { |
| 402 | return fmt.Errorf("slides: bump deck %s seq: %w", d.Sqid, res.Error) |
| 403 | } |
| 404 | if res.RowsAffected != 1 { |
| 405 | return errDeckBusy |
| 406 | } |
| 407 | return nil |
| 408 | }) |
| 409 | if err != nil { |
| 410 | return 0, err |
| 411 | } |
| 412 | d.Seq = next |
| 413 | return next, nil |
| 414 | } |
| 415 | |
| 416 | // mirrorMeta makes the deck row follow the document's meta after an |
| 417 | // update posted over the wire: the editor's Deck section writes |
| 418 | // meta.title and meta.theme this way, and the deck list, the settings |
| 419 | // form and every link's label read the row. |
| 420 | // |
| 421 | // It runs after the commit, deliberately, and it is the one place the |
| 422 | // wire path differs from the API's. metaRowUpdates can read a BATCH — |
| 423 | // it knows the ops — and so folds the row write into applyOps' own |
| 424 | // transaction. An update is bytes, not ops, so the only way to know |
| 425 | // what it did is to read the document back; previewUpdate already did, |
| 426 | // before the transaction, and v is that reading. Nothing here calls |
| 427 | // into the wasm module or holds SQLite's single writer for longer than |
| 428 | // one UPDATE. |
| 429 | // |
| 430 | // What that costs is honest: the row can be a moment behind the |
| 431 | // document, and a failure leaves it behind until the next update. Both |
| 432 | // are fine for what the row is — a label and a filter — and neither is |
| 433 | // data loss: the document is committed either way. |
| 434 | // |
| 435 | // seq is the seq this update landed at, and it is the guard: the write |
| 436 | // only lands while the row is still at it. Two updates racing would |
| 437 | // otherwise let the older one's read win, and the list would show a |
| 438 | // title nobody typed last. |
| 439 | func (a *app) mirrorMeta(ctx context.Context, d *Deck, v *deck.Deck, seq int64) { |
| 440 | changes := metaRowChanges(v, *d) |
| 441 | if len(changes) == 0 { |
| 442 | return |
| 443 | } |
| 444 | res := a.db.G.WithContext(ctx).Model(&Deck{}). |
| 445 | Where("id = ? AND seq = ?", d.ID, seq).Updates(changes) |
| 446 | if res.Error != nil { |
| 447 | a.logger.Error("slides: mirror a deck's meta onto its row", "deck", d.Sqid, "err", res.Error) |
| 448 | return |
| 449 | } |
| 450 | if res.RowsAffected != 1 { |
| 451 | return |
| 452 | } |
| 453 | if title, ok := changes["title"].(string); ok { |
| 454 | d.Title = title |
| 455 | } |
| 456 | if theme, ok := changes["theme"].(string); ok { |
| 457 | d.Theme = theme |
| 458 | } |
| 459 | } |
| 460 | |
| 461 | // metaRowChanges is the row's share of a document's meta: the columns |
| 462 | // that differ from what the row already holds, and nothing else — nil |
| 463 | // when nothing does, which is the answer for the vast majority of |
| 464 | // updates and the one that saves the write. |
| 465 | // |
| 466 | // An empty title is not a title: the document may legitimately have |
| 467 | // none, and blanking the row would leave a deck with no name anywhere |
| 468 | // a name is read. A theme the row cannot use is left out for |
| 469 | // metaRowUpdates' reason — the core takes any string, the row is what |
| 470 | // themes a render. |
| 471 | func metaRowChanges(v *deck.Deck, d Deck) map[string]any { |
| 472 | changes := map[string]any{} |
| 473 | if v.Meta.Title != "" && v.Meta.Title != d.Title { |
| 474 | changes["title"] = v.Meta.Title |
| 475 | } |
| 476 | if validTheme(v.Meta.Theme) && v.Meta.Theme != d.Theme { |
| 477 | changes["theme"] = v.Meta.Theme |
| 478 | } |
| 479 | if len(changes) == 0 { |
| 480 | return nil |
| 481 | } |
| 482 | return changes |
| 483 | } |
| 484 | |
| 485 | // --- GET /d/{sqid}/stream --- |
| 486 | |
| 487 | // deckStream is the deck's SSE feed: `event: update` carrying a seq and |
| 488 | // no bytes, `event: presence` carrying the whole set as JSON, a comment |
| 489 | // ping in between, and one of each on connect. |
| 490 | // |
| 491 | // The timings are presentation.go's, shared rather than restated |
| 492 | // (streamHeartbeat, streamWriteDeadline, streamTTL): the reasons are |
| 493 | // the same reasons, and two apps' worth of streams in one binary |
| 494 | // behaving differently would be a surprise nobody asked for. |
| 495 | // |
| 496 | // The connect events are load-bearing rather than an optimisation. The |
| 497 | // update one heals a tab whose cursor legitimately failed to advance |
| 498 | // and covers the gap between the page rendering and this stream |
| 499 | // opening; the presence one tells a tab who is already here, rather |
| 500 | // than leaving it to find out as each of them happens to move. |
| 501 | func (a *app) deckStream(w http.ResponseWriter, r *http.Request) { |
| 502 | u, ok := a.viewer(w, r) |
| 503 | if !ok { |
| 504 | return |
| 505 | } |
| 506 | d, ok := a.deckFor(w, r, u) |
| 507 | if !ok { |
| 508 | return |
| 509 | } |
| 510 | fl, ok := w.(http.Flusher) |
| 511 | if !ok { |
| 512 | a.fail(w, r, "stream: the ResponseWriter does not flush", nil) |
| 513 | return |
| 514 | } |
| 515 | |
| 516 | // Subscribe BEFORE re-reading the seq, never after (Docs' note): an |
| 517 | // update committing between the two must produce a redundant |
| 518 | // notification, which costs one empty fetch, where the other order |
| 519 | // would let it fall into the gap and leave the tab looking current |
| 520 | // and being stale until somebody typed. |
| 521 | ch, stop := a.live.Subscribe(d.Sqid) |
| 522 | defer stop() |
| 523 | |
| 524 | var current Deck |
| 525 | if err := a.db.G.WithContext(r.Context()).Take(¤t, d.ID).Error; err != nil { |
| 526 | a.fail(w, r, "read a deck for the stream", err) |
| 527 | return |
| 528 | } |
| 529 | |
| 530 | rc := http.NewResponseController(w) |
| 531 | w.Header().Set("Content-Type", "text/event-stream") |
| 532 | w.Header().Set("Cache-Control", "no-store") |
| 533 | w.WriteHeader(http.StatusOK) |
| 534 | fl.Flush() |
| 535 | |
| 536 | // write is every byte's one exit: deadline armed, bytes, flush. The |
| 537 | // deadline error is ignored deliberately — a ResponseWriter that |
| 538 | // cannot set deadlines still streams, unbounded — while a write |
| 539 | // error ends the handler, because it means the client is gone. |
| 540 | write := func(text string) bool { |
| 541 | _ = rc.SetWriteDeadline(time.Now().Add(streamWriteDeadline)) |
| 542 | if _, err := io.WriteString(w, text); err != nil { |
| 543 | return false |
| 544 | } |
| 545 | fl.Flush() |
| 546 | return true |
| 547 | } |
| 548 | seqEvent := func(seq int64) bool { |
| 549 | return write("event: update\ndata: " + strconv.FormatInt(seq, 10) + "\n\n") |
| 550 | } |
| 551 | presenceEvent := func(set live.PresenceSet) bool { |
| 552 | // A nil set must go out as `[]`, not `null`: the client reads |
| 553 | // this as "everybody here", and null is a different sentence. |
| 554 | if set == nil { |
| 555 | set = live.PresenceSet{} |
| 556 | } |
| 557 | body, err := json.Marshal(set) |
| 558 | if err != nil { |
| 559 | a.logger.Error("slides: encode presence", "deck", d.Sqid, "err", err) |
| 560 | return false |
| 561 | } |
| 562 | return write("event: presence\ndata: " + string(body) + "\n\n") |
| 563 | } |
| 564 | |
| 565 | if !seqEvent(current.Seq) { |
| 566 | return |
| 567 | } |
| 568 | if !presenceEvent(a.live.Snapshot(d.Sqid)) { |
| 569 | return |
| 570 | } |
| 571 | |
| 572 | heartbeat := time.NewTimer(streamHeartbeat) |
| 573 | defer heartbeat.Stop() |
| 574 | ttl := time.NewTimer(streamTTL) |
| 575 | defer ttl.Stop() |
| 576 | |
| 577 | for { |
| 578 | select { |
| 579 | case <-r.Context().Done(): |
| 580 | // The client went away, or the server is draining. Either |
| 581 | // way this returns, so a graceful shutdown's budget is not |
| 582 | // held open by a parked tab. |
| 583 | return |
| 584 | case <-ttl.C: |
| 585 | // A clean close: the browser reopens if it still wants one, |
| 586 | // and a tab nobody is looking at does not. |
| 587 | return |
| 588 | case e, ok := <-ch: |
| 589 | if !ok { |
| 590 | // Dropped for falling behind. Ending the response is |
| 591 | // the right answer: EventSource reconnects, and the |
| 592 | // connect events make it right again immediately. |
| 593 | return |
| 594 | } |
| 595 | sent := false |
| 596 | if e.Presence != nil { |
| 597 | sent = presenceEvent(*e.Presence) |
| 598 | } else { |
| 599 | sent = seqEvent(e.Seq) |
| 600 | } |
| 601 | if !sent { |
| 602 | return |
| 603 | } |
| 604 | heartbeat.Reset(streamHeartbeat) |
| 605 | case <-heartbeat.C: |
| 606 | if !write(": ping\n\n") { |
| 607 | return |
| 608 | } |
| 609 | heartbeat.Reset(streamHeartbeat) |
| 610 | } |
| 611 | } |
| 612 | } |
| 613 | |
| 614 | // --- POST /d/{sqid}/presence --- |
| 615 | |
| 616 | // deckPresence is one tab saying "I am here, on this slide, in this |
| 617 | // block". The client posts every twenty seconds and on every change |
| 618 | // (spec §3.3); two missed beats and the hub's sweeper drops it. |
| 619 | // |
| 620 | // Name and Colour are stamped from the users row and are NOT read from |
| 621 | // the body. Anything else would let anybody signed in appear in a deck |
| 622 | // wearing a colleague's name, which is the whole value of presence |
| 623 | // inverted. |
| 624 | // |
| 625 | // Nothing here is persisted. Presence is not document state: it never |
| 626 | // reaches the update log, it is never snapshotted, and it is gone when |
| 627 | // this process is. |
| 628 | func (a *app) deckPresence(w http.ResponseWriter, r *http.Request) { |
| 629 | u, ok := a.viewer(w, r) |
| 630 | if !ok { |
| 631 | return |
| 632 | } |
| 633 | d, ok := a.deckFor(w, r, u) |
| 634 | if !ok { |
| 635 | return |
| 636 | } |
| 637 | |
| 638 | r.Body = http.MaxBytesReader(w, r.Body, maxPresenceBytes) |
| 639 | var body struct { |
| 640 | Client string `json:"client"` |
| 641 | Slide string `json:"slide"` |
| 642 | Block string `json:"block"` |
| 643 | } |
| 644 | if err := json.NewDecoder(r.Body).Decode(&body); err != nil { |
| 645 | var tooBig *http.MaxBytesError |
| 646 | if errors.As(err, &tooBig) { |
| 647 | http.Error(w, rastrillo.T(r, "slides.live.error.toolarge"), http.StatusRequestEntityTooLarge) |
| 648 | return |
| 649 | } |
| 650 | liveInvalid(w, r) |
| 651 | return |
| 652 | } |
| 653 | // A heartbeat with no client to key on would be filed under the |
| 654 | // empty string, where every tab in the deck would overwrite every |
| 655 | // other. Refused rather than guessed at. |
| 656 | if body.Client == "" || len(body.Client) > maxPresenceField || |
| 657 | len(body.Slide) > maxPresenceField || len(body.Block) > maxPresenceField { |
| 658 | liveInvalid(w, r) |
| 659 | return |
| 660 | } |
| 661 | |
| 662 | name := u.Name |
| 663 | if name == "" { |
| 664 | name = u.Email |
| 665 | } |
| 666 | a.live.PublishPresence(d.Sqid, live.Presence{ |
| 667 | Client: body.Client, |
| 668 | Name: name, |
| 669 | Colour: u.Colour, |
| 670 | Slide: body.Slide, |
| 671 | Block: body.Block, |
| 672 | At: time.Now().UTC(), |
| 673 | }) |
| 674 | w.WriteHeader(http.StatusNoContent) |
| 675 | } |
| 676 | |
| 677 | // liveInvalid is the 422 a JSON caller on this surface gets: a body |
| 678 | // shaped like every other JSON error this app writes, so a client has |
| 679 | // one shape to read rather than two. The binary update endpoint answers |
| 680 | // the same status as plain text, because its caller is posting bytes |
| 681 | // and has no JSON decoder in the path at all. |
| 682 | func liveInvalid(w http.ResponseWriter, r *http.Request) { |
| 683 | w.Header().Set("Content-Type", "application/json; charset=utf-8") |
| 684 | w.WriteHeader(http.StatusUnprocessableEntity) |
| 685 | _ = json.NewEncoder(w).Encode(map[string]string{ |
| 686 | "error": "invalid", |
| 687 | "message": rastrillo.T(r, "slides.live.error.invalid"), |
| 688 | }) |
| 689 | } |
| 690 | |
amadan