amadan
feat: live-sync updates, stream, presence and snapshots
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01563XGCgxcSZToEXwuTBsKE
15 files changed,
+2077
−38
AGENTS.md+28 −15internal/slides/app.go+16 −0internal/slides/decks.go+26 −18internal/slides/editor.go+96 −0internal/slides/handlers.go+20 −5internal/slides/live.go+561 −0internal/slides/live/hub.go+319 −0internal/slides/live/hub_test.go+168 −0internal/slides/locales/en.toml+10 −0internal/slides/migrations/0006_create_deck_snapshots.sql+4 −0internal/slides/migrations/schema.sql+4 −0internal/slides/models.go+37 −0internal/slides/snapshot.go+152 −0internal/slides/snapshot_test.go+125 −0internal/slidestest/live_test.go+511 −0
Intent
Attributed to this commit: 8.69 g CO₂e (1033641 tokens). This is an attribution, not a measurement — usage counted from this commit's own marker forward to the next one on slice-2-editor, never backward from it. A turn that produced more than one commit books all of its tokens to the last one, and a concurrent session on this branch can add to the same figure. Some of it was priced using the shared default rate, because no model-specific factor row exists yet for at least one model involved. The Carbon view has the repo total, which is never derived from this attribution.
diff --git a/AGENTS.md b/AGENTS.md| index fd18a60..f75a9b3 100644 |
| --- a/AGENTS.md |
| +++ b/AGENTS.md |
| @@ -74,10 +74,11 @@ person and an agent both. |
| `internal/slides/` is the app in SKILL.md's five-file shape — `models.go`, |
| `migrations.go` plus `migrations/`, `app.go`, `handlers.go`, `render.go` |
| plus `templates/` — with one file per screen area beside them (`auth.go`, |
| -`decks.go`, `viewer.go`, `share.go`, `presentation.go`, `sample.go`) and |
| -one package per concern nested underneath: `core` (the wazero host), |
| -`deck` (Go types for the core's JSON view), `render` (slide HTML), |
| -`present` (the SSE hub). `core/` at the repo root is the Rust crate. |
| +`decks.go`, `viewer.go`, `share.go`, `presentation.go`, `sample.go`, |
| +`live.go`, `editor.go`, `snapshot.go`) and one package per concern nested |
| +underneath: `core` (the wazero host), `deck` (Go types for the core's JSON |
| +view), `render` (slide HTML), `present` (the presentation SSE hub), `live` |
| +(the deck's own hub: seqs and presence). `core/` at the repo root is the Rust crate. |
| `manifest/` and `gen/` are rastrillo's declarative path, unused here. |
| ## The seams |
| @@ -100,17 +101,23 @@ operator precedence. `deckFor` is a deck the person may read; `ownedDeckFor` |
| is one they own. Both answer **404, never 403**, so a stranger holding a |
| sqid cannot learn whether it is real. |
| -`openDeck` instantiates the Y.Doc: a fresh wasm module, every stored update |
| -replayed in seq order, and the JSON view it produces. The handle is the |
| -caller's to close. There are no snapshots, so this replays the whole log |
| -every time — fine at today's sizes, and the thing to fix somewhere around a |
| -thousand updates on one deck. |
| - |
| -`applyOps` is the only write path into a document. It puts the core's |
| -update row, the `seq` bump and any deck-row columns in one transaction with |
| -the seq as an optimistic lock, so the row and the log commit or roll back |
| -together. `editDeck` wraps it with the handle's whole life and one retry on |
| -`errDeckBusy`. Nothing else may append to `deck_updates`. |
| +`openDeck` instantiates the Y.Doc: a fresh wasm module, the newest |
| +`deck_snapshots` row applied, every update after it replayed in seq order, |
| +and the JSON view that produces. The handle is the caller's to close. A |
| +snapshot is derived data and never the truth (`snapshot.go`) — the log |
| +alone still is the deck, so a missing or unreadable snapshot means a slower |
| +open and nothing else. |
| + |
| +`applyOps` and `appendUpdate` are the two write paths into a document, and |
| +the only two. Each puts the update row, the `seq` bump and any deck-row |
| +columns in one transaction with the seq as an optimistic lock, so the row |
| +and the log commit or roll back together, then publishes the new seq and |
| +takes a snapshot if it crossed the boundary. They differ in one line and in |
| +what a failure of it means: `applyOps` (`editDeck` wraps it with the |
| +handle's life and one retry on `errDeckBusy`) hands the core an op batch |
| +this app wrote, so a refusal is a bug; `appendUpdate` (`live.go`) hands it |
| +bytes off the wire, so a refusal is ordinary and comes back as a 422. |
| +Nothing else may append to `deck_updates`. |
| `render.Slide(d, index, viewport)` is the only place slide HTML is |
| produced, for every surface. It builds markup by hand rather than |
| @@ -123,6 +130,12 @@ are whole state, never deltas, so a subscriber that falls behind is dropped |
| and reconnects rather than having events skipped — a skipped event leaves a |
| projector showing slide 4 for the rest of a talk with nothing to notice it. |
| +`live.Hub` is the same shape for a deck being edited: a seq, which is |
| +advisory ("there is something at least this new" — the bytes come from |
| +`GET /d/{sqid}/updates?since=`), and the whole presence set, which is not. |
| +Presence lives there and nowhere else — it is not document state, so it |
| +never reaches the log, is never snapshotted, and is gone with the process. |
| + |
| ## Traps |
| - **Never call `sessions.UserID`.** Under rastrillo/auth the Subject is |
diff --git a/internal/slides/app.go b/internal/slides/app.go| index 426e519..a819f17 100644 |
| --- a/internal/slides/app.go |
| +++ b/internal/slides/app.go |
| @@ -12,6 +12,7 @@ import ( |
| "amadan.net/rastrillo/rastrillo/migrate" |
| "amadan.net/oficina/slides/internal/slides/core" |
| + "amadan.net/oficina/slides/internal/slides/live" |
| "amadan.net/oficina/slides/internal/slides/present" |
| ) |
| @@ -99,6 +100,20 @@ func New(d *db.DB, cfg Config) (*http.ServeMux, *app, error) { |
| // Every slide's notes on one page — the crib sheet somebody |
| // presents from with scripts off (spec §8). |
| r.Get("/d/{sqid}/notes", a.deckNotes) |
| + // Live sync (live.go, spec §3.3). All four are the deck's |
| + // readers-and-writers — deckFor's audience, not the owner's |
| + // alone — because a team deck is edited by the team. |
| + // |
| + // Inside the group and therefore inside csrf.Protect, unlike |
| + // the drive API: these are made by a browser, which sends the |
| + // same-origin evidence csrf.Protect looks for. |
| + r.Get("/d/{sqid}/updates", a.updatesGet) |
| + r.Post("/d/{sqid}/updates", a.updatesPost) |
| + r.Get("/d/{sqid}/stream", a.deckStream) |
| + r.Post("/d/{sqid}/presence", a.deckPresence) |
| + // The slide HTML the editor's stage and strip fetch — the |
| + // session-cookie twin of the API's own (editor.go). |
| + r.Get("/d/{sqid}/slides/{file}", a.deckSlide) |
| // Presenting starts here and nowhere else: anybody who can |
| // open the deck can read it to a room, so this is deckFor's |
| // audience rather than the owner's alone. |
| @@ -213,5 +228,6 @@ func newApp(d *db.DB, cfg Config) (*app, *rauth.Auth, error) { |
| core: rt, |
| auth: authPlugin, |
| hub: present.New(), |
| + live: live.New(), |
| }, authPlugin, nil |
| } |
diff --git a/internal/slides/decks.go b/internal/slides/decks.go| index 71dd2b7..851f092 100644 |
| --- a/internal/slides/decks.go |
| +++ b/internal/slides/decks.go |
| @@ -178,28 +178,33 @@ func (a *app) visibleDecks(ctx context.Context, viewer User) *gorm.DB { |
| // --- the document --- |
| -// openDeck instantiates d's Y.Doc: a fresh wasm module, every stored |
| -// update replayed into it in seq order, and the view it produces. |
| +// openDeck instantiates d's Y.Doc: a fresh wasm module, the newest |
| +// snapshot applied, every update after it replayed in seq order, and |
| +// the view that produces. |
| // |
| // The handle is the CALLER's to close. One module per request is |
| -// deliberate — an instance is not safe for concurrent use, and task 6's |
| -// hub is where a long-lived one will live. |
| +// deliberate — an instance is not safe for concurrent use, and a |
| +// long-lived one is a change to make when there is evidence for it. |
| // |
| -// There are no snapshots yet: this replays the whole log every time. |
| -// That is fine at the sizes this app has today and stops being fine |
| -// somewhere around a thousand updates on one deck, which is the moment |
| -// to persist Handle.State as a snapshot row and replay from it. |
| +// The snapshot is an accelerator and never the truth (snapshot.go): a |
| +// deck with none, or one whose snapshot row could not be read, opens |
| +// from the plain log and is exactly the same deck, only slower. |
| func (a *app) openDeck(ctx context.Context, d Deck) (*core.Handle, *deck.Deck, error) { |
| h, err := a.core.NewDoc(ctx, d.Sqid, serverClientID) |
| if err != nil { |
| return nil, nil, fmt.Errorf("slides: open deck %s: %w", d.Sqid, err) |
| } |
| - var updates []DeckUpdate |
| - if err := a.db.G.WithContext(ctx). |
| - Where("deck_id = ?", d.ID).Order("seq").Find(&updates).Error; err != nil { |
| + snap, updates, err := a.replayPlan(ctx, d) |
| + if err != nil { |
| h.Close(ctx) |
| - return nil, nil, fmt.Errorf("slides: read deck %s log: %w", d.Sqid, err) |
| + return nil, nil, err |
| + } |
| + if snap != nil { |
| + if err := h.Apply(ctx, snap.Bytes); err != nil { |
| + h.Close(ctx) |
| + return nil, nil, fmt.Errorf("slides: replay deck %s snapshot at seq %d: %w", d.Sqid, snap.Seq, err) |
| + } |
| } |
| for _, u := range updates { |
| if err := h.Apply(ctx, u.Bytes); err != nil { |
| @@ -323,12 +328,15 @@ func (a *app) applyOps(ctx context.Context, h *core.Handle, d *Deck, actor strin |
| } |
| d.Seq++ |
| - // After the commit, never inside it: a subscriber told about a seq |
| - // a rolled-back transaction never wrote would broadcast an update |
| - // nobody can read back. Nil until task 6 mounts the hub. |
| - if a.onUpdate != nil { |
| - a.onUpdate(d.ID, d.Seq) |
| - } |
| + // After the commit, never inside it: a browser told about a seq a |
| + // rolled-back transaction never wrote would fetch it and find |
| + // nothing, and its cursor would sit one ahead of the deck for the |
| + // rest of the session. |
| + a.live.PublishSeq(d.Sqid, d.Seq) |
| + // Also after the commit, and best effort: h now holds exactly the |
| + // document through d.Seq, which is the one thing maybeSnapshot |
| + // cannot check for itself. |
| + a.maybeSnapshot(ctx, *d, h, d.Seq) |
| return view, nil |
| } |
diff --git a/internal/slides/editor.go b/internal/slides/editor.go| new file mode 100644 |
| index 0000000..e3442e9 |
| --- /dev/null |
| +++ b/internal/slides/editor.go |
| @@ -0,0 +1,96 @@ |
| +// editor.go is the browser editor's own surface. Today it holds one |
| +// handler: the slide HTML the stage and the strip fetch. The editor |
| +// page itself lands beside it. |
| +package slides |
| + |
| +import ( |
| + "fmt" |
| + "net/http" |
| + "strconv" |
| + "strings" |
| + |
| + "github.com/go-chi/chi/v5" |
| + |
| + slidehtml "amadan.net/oficina/slides/internal/slides/render" |
| +) |
| + |
| +// --- GET /d/{sqid}/slides/{id}.html --- |
| + |
| +// deckSlide renders one slide as the section the viewer would show it |
| +// in — the session-cookie twin of the API's apiSlide, for the editor's |
| +// stage and its thumbnail strip. |
| +// |
| +// Two surfaces rather than one shared route because the two are |
| +// authenticated differently and nothing else about them differs: the |
| +// API's caller carries a bearer token and the API's own 404 shape, this |
| +// one carries a session cookie and this app's. The rendering is |
| +// literally the same call into the same package, which is the invariant |
| +// that matters (AGENTS.md: render.Slide is the only place slide HTML is |
| +// produced). |
| +// |
| +// It is the SECTION, never a page: the stage swaps this into the DOM, |
| +// so a layout around it would be a page nested inside a page. |
| +// |
| +// vp names the viewport: `phone` reflows (spec §2.5), `desktop` and |
| +// `thumb` both render desktop. A thumbnail is a desktop slide scaled by |
| +// CSS rather than a third rendering — the one rule this app has is that |
| +// a slide is one block tree rendered many ways, and "many ways" is two |
| +// viewports, not one per place a slide is drawn. |
| +func (a *app) deckSlide(w http.ResponseWriter, r *http.Request) { |
| + u, ok := a.viewer(w, r) |
| + if !ok { |
| + return |
| + } |
| + d, ok := a.deckFor(w, r, u) |
| + if !ok { |
| + return |
| + } |
| + slideID, ok := strings.CutSuffix(chi.URLParam(r, "file"), ".html") |
| + if !ok || slideID == "" { |
| + a.notFound(w, r) |
| + return |
| + } |
| + |
| + h, view, err := a.openDeck(r.Context(), d) |
| + if err != nil { |
| + a.fail(w, r, "open a deck to render a slide", err) |
| + return |
| + } |
| + defer h.Close(r.Context()) |
| + |
| + index := -1 |
| + for i, s := range view.Slides { |
| + if s.ID == slideID { |
| + index = i |
| + break |
| + } |
| + } |
| + if index < 0 { |
| + a.notFound(w, r) |
| + return |
| + } |
| + |
| + vp := editorViewport(r.URL.Query().Get("vp")) |
| + section, err := slidehtml.Slide(view, index, vp) |
| + if err != nil { |
| + a.fail(w, r, "render a slide for the editor", err) |
| + return |
| + } |
| + w.Header().Set("Content-Type", "text/html; charset=utf-8") |
| + // Seq, slide and viewport: the three things that change what these |
| + // bytes are. The stage re-fetches after every structural change, so |
| + // this is what makes the fetch that changed nothing cheap. |
| + w.Header().Set("ETag", strconv.Quote(strconv.FormatInt(d.Seq, 10)+"-"+slideID+"-"+vp.String())) |
| + w.WriteHeader(http.StatusOK) |
| + fmt.Fprint(w, section) |
| +} |
| + |
| +// editorViewport maps the ?vp= value onto a rendering. Anything |
| +// unrecognised is desktop rather than a 400: this is a rendering hint |
| +// on a read, and a stage that got a typo should show the slide. |
| +func editorViewport(vp string) slidehtml.Viewport { |
| + if vp == "phone" { |
| + return slidehtml.Phone |
| + } |
| + return slidehtml.Desktop |
| +} |
diff --git a/internal/slides/handlers.go b/internal/slides/handlers.go| index fb34a49..f9662d5 100644 |
| --- a/internal/slides/handlers.go |
| +++ b/internal/slides/handlers.go |
| @@ -8,6 +8,7 @@ import ( |
| "amadan.net/rastrillo/rastrillo/db" |
| "amadan.net/oficina/slides/internal/slides/core" |
| + "amadan.net/oficina/slides/internal/slides/live" |
| "amadan.net/oficina/slides/internal/slides/present" |
| ) |
| @@ -44,10 +45,14 @@ type app struct { |
| // One per app, because a subscriber list nobody else can reach is |
| // not a fan-out. |
| hub *present.Hub |
| - // onUpdate is told, after the transaction commits, that a deck |
| - // moved to a new seq. Nil until task 6 mounts the hub that |
| - // broadcasts to open browsers; every caller is nil-safe. |
| - onUpdate func(deckID, seq int64) |
| + // live is the in-process fan-out behind every deck being EDITED: |
| + // one accepted update announced as a seq, plus who is currently in |
| + // the deck (internal/slides/live). Separate from hub above because |
| + // they are separate subjects with separate lifetimes — a talk being |
| + // given and a document being written — and folding them into one |
| + // map keyed by two kinds of sqid would make every publish a |
| + // question about which kind it was. |
| + live *live.Hub |
| } |
| // Close releases the compiled deck core's wazero runtime — the one |
| @@ -62,7 +67,17 @@ type app struct { |
| // cmd/slides/main.go's shutdown path and every test harness call it so |
| // a compiled module is not leaked per app built. |
| func (a *app) Close() error { |
| - if a == nil || a.core == nil { |
| + if a == nil { |
| + return nil |
| + } |
| + // The live hub's sweeper is a goroutine, so it is stopped here for |
| + // the same reason the runtime is closed here: a test binary builds |
| + // one app per test, and a resource nobody releases is a leak per |
| + // app rather than per process. |
| + if a.live != nil { |
| + a.live.Close() |
| + } |
| + if a.core == nil { |
| return nil |
| } |
| return a.core.Close(context.Background()) |
diff --git a/internal/slides/live.go b/internal/slides/live.go| new file mode 100644 |
| index 0000000..1bd40a2 |
| --- /dev/null |
| +++ b/internal/slides/live.go |
| @@ -0,0 +1,561 @@ |
| +// live.go is the transport a browser edits a deck over: the update log |
| +// by cursor, one new update in, an advisory stream out, and a heartbeat |
| +// saying who is in the deck. |
| +// |
| +// It is Docs' live-sync design adopted whole (its internal/docs/{ |
| +// handlers.go, stream.go}), with one addition and one difference. |
| +// |
| +// The addition is presence, which is NOT document state (spec §3.3): it |
| +// lives in the hub's memory, never in the log, and is gone when this |
| +// process is. |
| +// |
| +// The difference is framing. Docs concatenates updates, because a Yjs |
| +// update is self-delimiting and its client feeds the whole blob to one |
| +// decoder. Slides frames each one as a 4-byte big-endian length then |
| +// the bytes, so the client can hand them to the core one at a time and |
| +// so a fixture can compare a frame against the update that produced it |
| +// (the plan's live-sync wire). Concatenation would make both of those |
| +// "decode and hope". |
| +// |
| +// What never travels on the stream is bytes. The stream says "there is |
| +// content through seq N" and the client fetches it through the endpoint |
| +// it already used at mount — one code path from stored update to a |
| +// browser's Y.Doc rather than two that can drift, and no base64 tax on |
| +// every keystroke batch. |
| +package slides |
| + |
| +import ( |
| + "context" |
| + "encoding/binary" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "net/http" |
| + "strconv" |
| + "time" |
| + |
| + "gorm.io/gorm" |
| + |
| + "amadan.net/rastrillo/rastrillo" |
| + |
| + "amadan.net/oficina/slides/internal/slides/core" |
| + "amadan.net/oficina/slides/internal/slides/live" |
| +) |
| + |
| +// maxUpdateBytes is the cap on one posted update (the plan's 1 MiB). |
| +// Generous for a keystroke batch and for a pasted slide; small enough |
| +// that a client cannot make this instance hold an arbitrary amount of |
| +// memory by opening one connection. |
| +const maxUpdateBytes = 1 << 20 |
| + |
| +// maxPresenceBytes is the cap on one heartbeat body. A heartbeat is |
| +// four short strings; anything approaching this is not one. |
| +const maxPresenceBytes = 4 << 10 |
| + |
| +// maxPresenceField bounds each string the client supplies. Ids in this |
| +// app are twelve characters (decks.go's newID) and a client id is a |
| +// browser's own random string, so 128 is roomy — and a cap of some kind |
| +// is required, because these strings are held in memory for the life of |
| +// a heartbeat by a request that only had to be signed in. |
| +const maxPresenceField = 128 |
| + |
| +// errUpdateRefused is the core looking at the bytes a client posted and |
| +// saying no: malformed wire bytes, or bytes that trapped the module and |
| +// were contained. It is the client's problem, so it is a 422 and an |
| +// Info-level log — accepting arbitrary input over HTTP and occasionally |
| +// being sent nonsense is the ordinary case, not an incident. |
| +// |
| +// Everything else out of appendUpdate — a database failure, a replay of |
| +// this deck's own history going wrong, a rebuild that itself failed — |
| +// is a 500, because the client did not cause it and an operator needs |
| +// to see it. |
| +var errUpdateRefused = errors.New("slides: the core refused this update") |
| + |
| +// --- GET /d/{sqid}/updates --- |
| + |
| +// updatesGet serves the deck's log from a cursor: `application/octet- |
| +// stream`, each update framed as a 4-byte big-endian length then the |
| +// bytes, in seq order. |
| +// |
| +// `Slides-Seq` is the highest seq in the BODY, or the cursor itself |
| +// when the body is empty — derived from the bytes in hand and never |
| +// from a second MAX(seq) query. The difference matters: a separate |
| +// query can see an update this response does not carry, and the client |
| +// would then record a cursor for bytes it never received and never ask |
| +// for them again. |
| +// |
| +// `Slides-Snapshot: <seq>` says the first frame is a snapshot rather |
| +// than a log row. A client that does not care can ignore it — a |
| +// snapshot is a Yjs update like any other and applying it is correct |
| +// either way — and one that does can tell "here is the deck" from "here |
| +// is what changed". |
| +func (a *app) updatesGet(w http.ResponseWriter, r *http.Request) { |
| + u, ok := a.viewer(w, r) |
| + if !ok { |
| + return |
| + } |
| + d, ok := a.deckFor(w, r, u) |
| + if !ok { |
| + return |
| + } |
| + since, ok := parseSince(w, r) |
| + if !ok { |
| + return |
| + } |
| + |
| + updates, highest, snapshotSeq, err := a.readUpdates(r.Context(), d, since) |
| + if err != nil { |
| + a.fail(w, r, "read a deck's log", err) |
| + return |
| + } |
| + |
| + w.Header().Set("Content-Type", "application/octet-stream") |
| + w.Header().Set("Cache-Control", "no-store") |
| + w.Header().Set("Slides-Seq", strconv.FormatInt(highest, 10)) |
| + if snapshotSeq > 0 { |
| + w.Header().Set("Slides-Snapshot", strconv.FormatInt(snapshotSeq, 10)) |
| + } |
| + |
| + var length [4]byte |
| + for _, u := range updates { |
| + binary.BigEndian.PutUint32(length[:], uint32(len(u))) |
| + if _, err := w.Write(length[:]); err != nil { |
| + // The client went away mid-response. Logged rather than |
| + // surfaced: the status line is already on the wire. |
| + a.logger.Error("slides: write the update log", "deck", d.Sqid, "err", err) |
| + return |
| + } |
| + if _, err := w.Write(u); err != nil { |
| + a.logger.Error("slides: write the update log", "deck", d.Sqid, "err", err) |
| + return |
| + } |
| + } |
| +} |
| + |
| +// readUpdates is what a cursor is owed: the frames to send, the highest |
| +// seq they carry, and the seq of the snapshot when one of them is one. |
| +// |
| +// The snapshot is only ever offered at since=0. A client mid-session |
| +// holds a prefix of the log and needs the tail after it; handing it a |
| +// whole-document update instead would be correct (Yjs would merge it) |
| +// and wasteful on every single fetch. At since=0 it is the opposite: a |
| +// deck with two thousand updates behind it would otherwise send all two |
| +// thousand to every tab that opens it. |
| +func (a *app) readUpdates(ctx context.Context, d Deck, since int64) (frames [][]byte, highest, snapshotSeq int64, err error) { |
| + highest = since |
| + tail := since |
| + |
| + if since == 0 { |
| + // An accelerator, never the source of truth (snapshot.go): a |
| + // lookup that fails serves the plain log rather than failing a |
| + // request the log alone can still answer. |
| + snap, err := a.latestSnapshot(ctx, d.ID) |
| + if err != nil { |
| + a.logger.Error("slides: load a deck's newest snapshot, serving the plain log instead", |
| + "deck", d.Sqid, "err", err) |
| + } else if snap != nil && snap.Seq > since { |
| + frames = append(frames, snap.Bytes) |
| + snapshotSeq = snap.Seq |
| + tail = snap.Seq |
| + highest = snap.Seq |
| + } |
| + } |
| + |
| + var updates []DeckUpdate |
| + if err := a.db.G.WithContext(ctx). |
| + Where("deck_id = ? AND seq > ?", d.ID, tail). |
| + Order("seq").Find(&updates).Error; err != nil { |
| + return nil, 0, 0, fmt.Errorf("slides: read deck %s log from %d: %w", d.Sqid, tail, err) |
| + } |
| + for _, u := range updates { |
| + frames = append(frames, u.Bytes) |
| + if u.Seq > highest { |
| + highest = u.Seq |
| + } |
| + } |
| + return frames, highest, snapshotSeq, nil |
| +} |
| + |
| +// parseSince reads the ?since= cursor: the highest seq the client |
| +// already holds, so the answer carries only what came after it. Absent |
| +// means 0, which asks for the whole deck — what a tab wants at mount. |
| +// |
| +// A malformed or negative value is a 400 rather than a silent 0. |
| +// Treating "banana" as "send everything" would answer an expensive |
| +// question nobody asked and hide a client bug behind a response that |
| +// looks fine. |
| +func parseSince(w http.ResponseWriter, r *http.Request) (int64, bool) { |
| + raw := r.URL.Query().Get("since") |
| + if raw == "" { |
| + return 0, true |
| + } |
| + since, err := strconv.ParseInt(raw, 10, 64) |
| + if err != nil || since < 0 { |
| + http.Error(w, rastrillo.T(r, "slides.live.error.badsince"), http.StatusBadRequest) |
| + return 0, false |
| + } |
| + return since, true |
| +} |
| + |
| +// --- POST /d/{sqid}/updates --- |
| + |
| +// updatesPost takes one Yjs update from a browser, applies it through |
| +// the core, stores it and answers 204 with the seq it landed at. |
| +// |
| +// The body is wrapped in MaxBytesReader before anything reads it, so an |
| +// oversized update is refused before it is buffered into memory at all, |
| +// never after. |
| +func (a *app) updatesPost(w http.ResponseWriter, r *http.Request) { |
| + u, ok := a.viewer(w, r) |
| + if !ok { |
| + return |
| + } |
| + d, ok := a.deckFor(w, r, u) |
| + if !ok { |
| + return |
| + } |
| + |
| + r.Body = http.MaxBytesReader(w, r.Body, maxUpdateBytes) |
| + update, err := io.ReadAll(r.Body) |
| + if err != nil { |
| + var tooBig *http.MaxBytesError |
| + if errors.As(err, &tooBig) { |
| + http.Error(w, rastrillo.T(r, "slides.live.error.toolarge"), http.StatusRequestEntityTooLarge) |
| + return |
| + } |
| + a.fail(w, r, "read a posted update", err) |
| + return |
| + } |
| + |
| + seq, err := a.appendUpdate(r.Context(), &d, actorFor(u), update) |
| + if err != nil { |
| + if errors.Is(err, errUpdateRefused) { |
| + a.logger.Info("slides: refuse a posted update", "deck", d.Sqid, "err", err) |
| + http.Error(w, rastrillo.T(r, "slides.live.error.invalid"), http.StatusUnprocessableEntity) |
| + return |
| + } |
| + if errors.Is(err, errDeckBusy) { |
| + // Two retries both lost: real contention rather than a |
| + // race worth spinning on. The client's own queue reposts, |
| + // and Yjs makes a duplicate harmless. |
| + http.Error(w, rastrillo.T(r, "slides.live.error.busy"), http.StatusConflict) |
| + return |
| + } |
| + a.fail(w, r, "apply a posted update", err) |
| + return |
| + } |
| + |
| + // The header before the broadcast, both only on success: an |
| + // announced seq must be one a reader can actually fetch back. |
| + w.Header().Set("Slides-Seq", strconv.FormatInt(seq, 10)) |
| + w.WriteHeader(http.StatusNoContent) |
| +} |
| + |
| +// appendUpdate is the second write path into a document, beside |
| +// applyOps, and it takes the same lock the same way: open the deck, |
| +// apply the update to the handle, insert the log row at Seq+1 and bump |
| +// the row's seq in ONE transaction whose `WHERE id = ? AND seq = ?` is |
| +// the optimistic lock. One retry on errDeckBusy from a freshly opened |
| +// document, for editDeck's reason — the batch applied nothing, so |
| +// replaying it against the state that beat us is correct. |
| +// |
| +// It exists rather than being folded into applyOps because the two |
| +// differ in exactly one line and in what a failure of that line MEANS. |
| +// applyOps hands the core a batch this app wrote, so a refusal is a bug |
| +// here; appendUpdate hands it bytes off the wire, so a refusal is the |
| +// ordinary case and has to come back distinguishable (errUpdateRefused) |
| +// rather than as a 500. |
| +// |
| +// d is a pointer because its Seq is updated in place on success and |
| +// re-read on a conflict, exactly as editDeck does. |
| +func (a *app) appendUpdate(ctx context.Context, d *Deck, actor string, update []byte) (int64, error) { |
| + var lastErr error |
| + for attempt := 0; attempt < 2; attempt++ { |
| + h, _, err := a.openDeck(ctx, *d) |
| + if err != nil { |
| + return 0, err |
| + } |
| + seq, err := a.commitUpdate(ctx, h, d, actor, update) |
| + if err == nil { |
| + // After the commit, in this order and for the same reasons |
| + // applyOps has: nobody is told about a seq a rolled-back |
| + // transaction never wrote, and the snapshot is taken from |
| + // a handle that holds exactly what the row now claims. |
| + a.live.PublishSeq(d.Sqid, seq) |
| + a.maybeSnapshot(ctx, *d, h, seq) |
| + h.Close(ctx) |
| + return seq, nil |
| + } |
| + h.Close(ctx) |
| + |
| + lastErr = err |
| + if !errors.Is(err, errDeckBusy) { |
| + return 0, err |
| + } |
| + // Re-read the row: our copy's Seq is what just lost. |
| + if err := a.db.G.WithContext(ctx).First(d, d.ID).Error; err != nil { |
| + return 0, fmt.Errorf("slides: reload deck %s: %w", d.Sqid, err) |
| + } |
| + } |
| + return 0, lastErr |
| +} |
| + |
| +// commitUpdate is appendUpdate's one attempt: applyOps' transaction |
| +// with h.Apply where h.Ops would be. |
| +// |
| +// The same honest wrinkle applies as in applyOps: Apply runs INSIDE the |
| +// transaction, so a rollback leaves the handle holding a change the |
| +// database does not. That is harmless because the handle is |
| +// per-attempt and discarded on failure, and it is why appendUpdate |
| +// reopens rather than retrying on the same handle. |
| +func (a *app) commitUpdate(ctx context.Context, h *core.Handle, d *Deck, actor string, update []byte) (int64, error) { |
| + next := d.Seq + 1 |
| + err := a.db.G.WithContext(ctx).Transaction(func(tx *gorm.DB) error { |
| + if err := h.Apply(ctx, update); err != nil { |
| + if core.Rejected(err) || core.Recovered(err) { |
| + return fmt.Errorf("%w: %w", errUpdateRefused, err) |
| + } |
| + // A rebuild that itself failed, or a full retained log: |
| + // neither is the client's doing, and the second means this |
| + // deck needs a snapshot before it can take anything else. |
| + return fmt.Errorf("slides: apply a posted update to deck %s: %w", d.Sqid, err) |
| + } |
| + |
| + if err := tx.Create(&DeckUpdate{ |
| + DeckID: d.ID, |
| + Seq: next, |
| + Actor: actor, |
| + Bytes: update, |
| + CreatedAt: time.Now().UTC(), |
| + }).Error; err != nil { |
| + if errors.Is(err, gorm.ErrDuplicatedKey) { |
| + // Somebody else's update is already at this seq: the |
| + // lost race, reported as the lost race. |
| + return errDeckBusy |
| + } |
| + return fmt.Errorf("slides: record deck %s update %d: %w", d.Sqid, next, err) |
| + } |
| + |
| + res := tx.Model(&Deck{}).Where("id = ? AND seq = ?", d.ID, d.Seq). |
| + Updates(map[string]any{"seq": next}) |
| + if res.Error != nil { |
| + return fmt.Errorf("slides: bump deck %s seq: %w", d.Sqid, res.Error) |
| + } |
| + if res.RowsAffected != 1 { |
| + return errDeckBusy |
| + } |
| + return nil |
| + }) |
| + if err != nil { |
| + return 0, err |
| + } |
| + d.Seq = next |
| + return next, nil |
| +} |
| + |
| +// --- GET /d/{sqid}/stream --- |
| + |
| +// deckStream is the deck's SSE feed: `event: update` carrying a seq and |
| +// no bytes, `event: presence` carrying the whole set as JSON, a comment |
| +// ping in between, and one of each on connect. |
| +// |
| +// The timings are presentation.go's, shared rather than restated |
| +// (streamHeartbeat, streamWriteDeadline, streamTTL): the reasons are |
| +// the same reasons, and two apps' worth of streams in one binary |
| +// behaving differently would be a surprise nobody asked for. |
| +// |
| +// The connect events are load-bearing rather than an optimisation. The |
| +// update one heals a tab whose cursor legitimately failed to advance |
| +// and covers the gap between the page rendering and this stream |
| +// opening; the presence one tells a tab who is already here, rather |
| +// than leaving it to find out as each of them happens to move. |
| +func (a *app) deckStream(w http.ResponseWriter, r *http.Request) { |
| + u, ok := a.viewer(w, r) |
| + if !ok { |
| + return |
| + } |
| + d, ok := a.deckFor(w, r, u) |
| + if !ok { |
| + return |
| + } |
| + fl, ok := w.(http.Flusher) |
| + if !ok { |
| + a.fail(w, r, "stream: the ResponseWriter does not flush", nil) |
| + return |
| + } |
| + |
| + // Subscribe BEFORE re-reading the seq, never after (Docs' note): an |
| + // update committing between the two must produce a redundant |
| + // notification, which costs one empty fetch, where the other order |
| + // would let it fall into the gap and leave the tab looking current |
| + // and being stale until somebody typed. |
| + ch, stop := a.live.Subscribe(d.Sqid) |
| + defer stop() |
| + |
| + var current Deck |
| + if err := a.db.G.WithContext(r.Context()).Take(¤t, d.ID).Error; err != nil { |
| + a.fail(w, r, "read a deck for the stream", err) |
| + return |
| + } |
| + |
| + rc := http.NewResponseController(w) |
| + w.Header().Set("Content-Type", "text/event-stream") |
| + w.Header().Set("Cache-Control", "no-store") |
| + w.WriteHeader(http.StatusOK) |
| + fl.Flush() |
| + |
| + // write is every byte's one exit: deadline armed, bytes, flush. The |
| + // deadline error is ignored deliberately — a ResponseWriter that |
| + // cannot set deadlines still streams, unbounded — while a write |
| + // error ends the handler, because it means the client is gone. |
| + write := func(text string) bool { |
| + _ = rc.SetWriteDeadline(time.Now().Add(streamWriteDeadline)) |
| + if _, err := io.WriteString(w, text); err != nil { |
| + return false |
| + } |
| + fl.Flush() |
| + return true |
| + } |
| + seqEvent := func(seq int64) bool { |
| + return write("event: update\ndata: " + strconv.FormatInt(seq, 10) + "\n\n") |
| + } |
| + presenceEvent := func(set live.PresenceSet) bool { |
| + // A nil set must go out as `[]`, not `null`: the client reads |
| + // this as "everybody here", and null is a different sentence. |
| + if set == nil { |
| + set = live.PresenceSet{} |
| + } |
| + body, err := json.Marshal(set) |
| + if err != nil { |
| + a.logger.Error("slides: encode presence", "deck", d.Sqid, "err", err) |
| + return false |
| + } |
| + return write("event: presence\ndata: " + string(body) + "\n\n") |
| + } |
| + |
| + if !seqEvent(current.Seq) { |
| + return |
| + } |
| + if !presenceEvent(a.live.Snapshot(d.Sqid)) { |
| + return |
| + } |
| + |
| + heartbeat := time.NewTimer(streamHeartbeat) |
| + defer heartbeat.Stop() |
| + ttl := time.NewTimer(streamTTL) |
| + defer ttl.Stop() |
| + |
| + for { |
| + select { |
| + case <-r.Context().Done(): |
| + // The client went away, or the server is draining. Either |
| + // way this returns, so a graceful shutdown's budget is not |
| + // held open by a parked tab. |
| + return |
| + case <-ttl.C: |
| + // A clean close: the browser reopens if it still wants one, |
| + // and a tab nobody is looking at does not. |
| + return |
| + case e, ok := <-ch: |
| + if !ok { |
| + // Dropped for falling behind. Ending the response is |
| + // the right answer: EventSource reconnects, and the |
| + // connect events make it right again immediately. |
| + return |
| + } |
| + sent := false |
| + if e.Presence != nil { |
| + sent = presenceEvent(*e.Presence) |
| + } else { |
| + sent = seqEvent(e.Seq) |
| + } |
| + if !sent { |
| + return |
| + } |
| + heartbeat.Reset(streamHeartbeat) |
| + case <-heartbeat.C: |
| + if !write(": ping\n\n") { |
| + return |
| + } |
| + heartbeat.Reset(streamHeartbeat) |
| + } |
| + } |
| +} |
| + |
| +// --- POST /d/{sqid}/presence --- |
| + |
| +// deckPresence is one tab saying "I am here, on this slide, in this |
| +// block". The client posts every twenty seconds and on every change |
| +// (spec §3.3); two missed beats and the hub's sweeper drops it. |
| +// |
| +// Name and Colour are stamped from the users row and are NOT read from |
| +// the body. Anything else would let anybody signed in appear in a deck |
| +// wearing a colleague's name, which is the whole value of presence |
| +// inverted. |
| +// |
| +// Nothing here is persisted. Presence is not document state: it never |
| +// reaches the update log, it is never snapshotted, and it is gone when |
| +// this process is. |
| +func (a *app) deckPresence(w http.ResponseWriter, r *http.Request) { |
| + u, ok := a.viewer(w, r) |
| + if !ok { |
| + return |
| + } |
| + d, ok := a.deckFor(w, r, u) |
| + if !ok { |
| + return |
| + } |
| + |
| + r.Body = http.MaxBytesReader(w, r.Body, maxPresenceBytes) |
| + var body struct { |
| + Client string `json:"client"` |
| + Slide string `json:"slide"` |
| + Block string `json:"block"` |
| + } |
| + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { |
| + var tooBig *http.MaxBytesError |
| + if errors.As(err, &tooBig) { |
| + http.Error(w, rastrillo.T(r, "slides.live.error.toolarge"), http.StatusRequestEntityTooLarge) |
| + return |
| + } |
| + liveInvalid(w, r) |
| + return |
| + } |
| + // A heartbeat with no client to key on would be filed under the |
| + // empty string, where every tab in the deck would overwrite every |
| + // other. Refused rather than guessed at. |
| + if body.Client == "" || len(body.Client) > maxPresenceField || |
| + len(body.Slide) > maxPresenceField || len(body.Block) > maxPresenceField { |
| + liveInvalid(w, r) |
| + return |
| + } |
| + |
| + name := u.Name |
| + if name == "" { |
| + name = u.Email |
| + } |
| + a.live.PublishPresence(d.Sqid, live.Presence{ |
| + Client: body.Client, |
| + Name: name, |
| + Colour: u.Colour, |
| + Slide: body.Slide, |
| + Block: body.Block, |
| + At: time.Now().UTC(), |
| + }) |
| + w.WriteHeader(http.StatusNoContent) |
| +} |
| + |
| +// liveInvalid is the 422 a JSON caller on this surface gets: a body |
| +// shaped like every other JSON error this app writes, so a client has |
| +// one shape to read rather than two. The binary update endpoint answers |
| +// the same status as plain text, because its caller is posting bytes |
| +// and has no JSON decoder in the path at all. |
| +func liveInvalid(w http.ResponseWriter, r *http.Request) { |
| + w.Header().Set("Content-Type", "application/json; charset=utf-8") |
| + w.WriteHeader(http.StatusUnprocessableEntity) |
| + _ = json.NewEncoder(w).Encode(map[string]string{ |
| + "error": "invalid", |
| + "message": rastrillo.T(r, "slides.live.error.invalid"), |
| + }) |
| +} |
diff --git a/internal/slides/live/hub.go b/internal/slides/live/hub.go| new file mode 100644 |
| index 0000000..4b4382f |
| --- /dev/null |
| +++ b/internal/slides/live/hub.go |
| @@ -0,0 +1,319 @@ |
| +// Package live is the fan-out behind a deck being edited: one accepted |
| +// update announced to every browser that has the deck open, and the set |
| +// of people currently in it. |
| +// |
| +// It is Docs' hub (its internal/docs/hub.go) with presence bolted on, |
| +// and the two things it carries are carried differently on purpose. |
| +// |
| +// A SEQ is advisory — "there is something at least this new". It never |
| +// says what changed, and a subscriber must not treat it as a cursor |
| +// assignment: the bytes come from GET /d/{sqid}/updates?since=<its own |
| +// cursor>, through the one replay path, so a subscriber that missed ten |
| +// notifications is still exactly right after one fetch. |
| +// |
| +// A PRESENCE event is the whole set for that deck, never a delta, for |
| +// present.Hub's reason: a subscriber that applies one is right rather |
| +// than approximately right, and there is no "fetch the presence I |
| +// missed" endpoint to make a skipped one safe. |
| +// |
| +// The two together decide the channel's shape. It is buffered eight — |
| +// a burst of keystroke batches is a real thing — and a subscriber that |
| +// falls behind that is DROPPED, channel closed, rather than having |
| +// events skipped. The handler holding it ends the response, the browser |
| +// reconnects, and its connect events re-read both the current seq and |
| +// the current set. Skipping would leave a stale cursor look correct. |
| +// |
| +// Presence lives here and nowhere else: it is not document state (spec |
| +// §3.3), so it is never written to the update log, never snapshotted, |
| +// and is gone when this process is. |
| +// |
| +// Scope, like Docs' and like present.Hub's: one process. Slides is one |
| +// instance per team, so there is no second process to reach; this type |
| +// is the seam that changes on the day there is. |
| +package live |
| + |
| +import ( |
| + "sort" |
| + "sync" |
| + "time" |
| +) |
| + |
| +// TTL is how long one person's presence survives without a heartbeat. |
| +// The client posts every 20 s and on every change (spec §3.3), so this |
| +// is two missed beats plus a little slack — long enough that a phone |
| +// switching networks does not flicker out of the deck, short enough |
| +// that somebody who closed their laptop stops being shown as present |
| +// while their colleague waits for a reply. |
| +const TTL = 45 * time.Second |
| + |
| +// sweepInterval is how often the sweeper looks for expired presences. |
| +// A third of TTL, so somebody who has gone is shown as gone within |
| +// about fifteen seconds of the deadline rather than at the next event |
| +// somebody else happens to cause. |
| +const sweepInterval = 15 * time.Second |
| + |
| +// subscriberBuffer is how many events a subscriber may fall behind by |
| +// before it is dropped. See the package comment for why it is dropped |
| +// rather than skipped. |
| +const subscriberBuffer = 8 |
| + |
| +// Presence is one browser tab in one deck at one moment. |
| +// |
| +// Client is the tab's own random string, minted per tab rather than |
| +// per person: one person with the deck open twice is two cursors, and |
| +// keying on the user id would make the second tab overwrite the first. |
| +// |
| +// Name and Colour are stamped by the SERVER from the users row, never |
| +// taken from the request body — otherwise anybody signed in could |
| +// appear in a deck under somebody else's name. |
| +// |
| +// At is when the last heartbeat landed, and it is what the sweeper |
| +// reads. It carries no JSON tag because it is bookkeeping: the browser |
| +// is told who is here, not when each of them last said so. |
| +type Presence struct { |
| + Client string `json:"client"` |
| + Name string `json:"name"` |
| + Colour string `json:"colour"` |
| + // Slide is the id of the slide this tab is looking at, and Block |
| + // the id of the block its caret is in. Both may be empty — a tab |
| + // that has just opened the deck is present without being anywhere |
| + // in particular yet. |
| + Slide string `json:"slide"` |
| + Block string `json:"block"` |
| + At time.Time `json:"-"` |
| +} |
| + |
| +// PresenceSet is everybody currently in one deck, ordered by Client so |
| +// two reads of an unchanged set produce byte-identical JSON — which is |
| +// what lets a client diff cheaply and what makes the tests here able to |
| +// say "the set did not change" at all. |
| +type PresenceSet []Presence |
| + |
| +// Event is what a subscriber receives. Exactly one of the two halves is |
| +// meaningful: a nil Presence makes it a seq announcement, a non-nil one |
| +// makes it a presence announcement whose Seq is not to be read. |
| +// |
| +// A pointer rather than a length check, because an EMPTY set is a real |
| +// event — it is what "the last person left" looks like — and a |
| +// nil-versus-empty slice distinction is exactly the kind that gets |
| +// flattened by an innocent refactor. |
| +type Event struct { |
| + Seq int64 |
| + Presence *PresenceSet |
| +} |
| + |
| +// Hub is the per-process fan-out, keyed by deck sqid. |
| +// |
| +// Keyed by sqid rather than by deck id because that is what a URL, a |
| +// handler and a browser all already hold; nothing here ever needs to |
| +// touch the database, so there is nothing to look an id up against. |
| +type Hub struct { |
| + mu sync.Mutex |
| + subs map[string]map[chan Event]struct{} |
| + // people is deck sqid -> client -> that client's last heartbeat. |
| + people map[string]map[string]Presence |
| + // now is the clock, swappable by this package's own tests so |
| + // expiry is asserted by moving time rather than by waiting for it. |
| + now func() time.Time |
| + |
| + stop chan struct{} |
| + stopOnce sync.Once |
| +} |
| + |
| +// New builds an empty hub and starts its sweeper. One per app (app.go). |
| +// |
| +// The sweeper is a goroutine rather than work done lazily on the next |
| +// event, because the event that matters most is the one nobody causes: |
| +// the last person in a deck closing their laptop. Lazily, their cursor |
| +// would sit on somebody else's screen until that person typed. |
| +// |
| +// Close stops it. Every caller that builds a Hub must, or a test binary |
| +// leaks one goroutine per app it builds. |
| +func New() *Hub { |
| + h := &Hub{ |
| + subs: make(map[string]map[chan Event]struct{}), |
| + people: make(map[string]map[string]Presence), |
| + now: time.Now, |
| + stop: make(chan struct{}), |
| + } |
| + go h.sweeper(sweepInterval) |
| + return h |
| +} |
| + |
| +// Close stops the sweeper. Idempotent, because the app that owns a hub |
| +// may be closed by both a shutdown path and a test cleanup. |
| +func (h *Hub) Close() { |
| + h.stopOnce.Do(func() { close(h.stop) }) |
| +} |
| + |
| +// Subscribe registers interest in one deck and returns the channel to |
| +// wait on plus the function that unregisters it. |
| +// |
| +// The returned cancel is idempotent — the stream handler defers it and |
| +// may also have reason to stop early — and it never closes the channel. |
| +// Only the publish path closes one, and only when it is dropping the |
| +// subscriber, so "the channel is closed" carries exactly one meaning |
| +// for a reader: you fell behind, reconnect. A cancel that closed too |
| +// would make that signal ambiguous at the one moment it matters. |
| +func (h *Hub) Subscribe(deck string) (<-chan Event, func()) { |
| + ch := make(chan Event, subscriberBuffer) |
| + |
| + h.mu.Lock() |
| + if h.subs[deck] == nil { |
| + h.subs[deck] = make(map[chan Event]struct{}) |
| + } |
| + h.subs[deck][ch] = struct{}{} |
| + h.mu.Unlock() |
| + |
| + var once sync.Once |
| + return ch, func() { |
| + once.Do(func() { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + h.removeLocked(deck, ch) |
| + }) |
| + } |
| +} |
| + |
| +// PublishSeq tells everyone watching deck that there is content through |
| +// seq. It runs on whichever request goroutine just committed an update, |
| +// so it never blocks. |
| +func (h *Hub) PublishSeq(deck string, seq int64) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + h.publishLocked(deck, Event{Seq: seq}) |
| +} |
| + |
| +// PublishPresence records one heartbeat and announces the deck's whole |
| +// set. A zero At is stamped here rather than refused, so a caller that |
| +// only has {client, slide, block} to hand cannot accidentally publish a |
| +// presence that the sweeper reads as forty-five seconds stale. |
| +func (h *Hub) PublishPresence(deck string, p Presence) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + |
| + if p.At.IsZero() { |
| + p.At = h.now() |
| + } |
| + if h.people[deck] == nil { |
| + h.people[deck] = make(map[string]Presence) |
| + } |
| + h.people[deck][p.Client] = p |
| + |
| + set := h.setLocked(deck) |
| + h.publishLocked(deck, Event{Presence: &set}) |
| +} |
| + |
| +// Snapshot is the deck's current set, for the stream's connect event — |
| +// so a tab that opens after everybody else already arrived sees them, |
| +// rather than waiting for each of them to move. |
| +// |
| +// Expired entries are filtered here as well as swept, because a read |
| +// must never be able to report somebody as present past their deadline |
| +// just because the sweeper's next tick has not come round yet. |
| +func (h *Hub) Snapshot(deck string) PresenceSet { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + return h.setLocked(deck) |
| +} |
| + |
| +// sweeper is the goroutine New starts. |
| +func (h *Hub) sweeper(every time.Duration) { |
| + t := time.NewTicker(every) |
| + defer t.Stop() |
| + for { |
| + select { |
| + case <-h.stop: |
| + return |
| + case <-t.C: |
| + h.sweep() |
| + } |
| + } |
| +} |
| + |
| +// sweep drops every presence past TTL and republishes the set for each |
| +// deck that actually changed — and only those, because an unchanged |
| +// deck republished every fifteen seconds would be a heartbeat pretending |
| +// to be an event, waking every open tab for nothing. |
| +func (h *Hub) sweep() { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + |
| + cut := h.now().Add(-TTL) |
| + for deck, set := range h.people { |
| + changed := false |
| + for client, p := range set { |
| + if p.At.Before(cut) { |
| + delete(set, client) |
| + changed = true |
| + } |
| + } |
| + if len(set) == 0 { |
| + // Drop the inner map with its last member, or this map |
| + // keeps one empty entry per deck ever opened for the life |
| + // of the process. |
| + delete(h.people, deck) |
| + } |
| + if !changed { |
| + continue |
| + } |
| + announced := h.setLocked(deck) |
| + h.publishLocked(deck, Event{Presence: &announced}) |
| + } |
| +} |
| + |
| +// setLocked builds deck's set: a sorted copy with expired entries left |
| +// out. Callers hold h.mu. |
| +func (h *Hub) setLocked(deck string) PresenceSet { |
| + cut := h.now().Add(-TTL) |
| + set := make(PresenceSet, 0, len(h.people[deck])) |
| + for _, p := range h.people[deck] { |
| + if p.At.Before(cut) { |
| + continue |
| + } |
| + set = append(set, p) |
| + } |
| + sort.Slice(set, func(i, j int) bool { return set[i].Client < set[j].Client }) |
| + return set |
| +} |
| + |
| +// publishLocked sends e to everyone watching deck. Callers hold h.mu, |
| +// which is what makes the close below safe: no send can race it. |
| +func (h *Hub) publishLocked(deck string, e Event) { |
| + for ch := range h.subs[deck] { |
| + select { |
| + case ch <- e: |
| + default: |
| + h.removeLocked(deck, ch) |
| + close(ch) |
| + } |
| + } |
| +} |
| + |
| +// removeLocked takes ch out of deck's set and drops the set with its |
| +// last member. Callers hold h.mu. |
| +func (h *Hub) removeLocked(deck string, ch chan Event) { |
| + subs, ok := h.subs[deck] |
| + if !ok { |
| + return |
| + } |
| + delete(subs, ch) |
| + if len(subs) == 0 { |
| + delete(h.subs, deck) |
| + } |
| +} |
| + |
| +// deckCount and presenceDecks report how many decks the two maps hold. |
| +// They exist for the tests that assert neither grows forever; nothing |
| +// in the app reads them. |
| +func (h *Hub) deckCount() int { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + return len(h.subs) |
| +} |
| + |
| +func (h *Hub) presenceDecks() int { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + return len(h.people) |
| +} |
diff --git a/internal/slides/live/hub_test.go b/internal/slides/live/hub_test.go| new file mode 100644 |
| index 0000000..4d6d9de |
| --- /dev/null |
| +++ b/internal/slides/live/hub_test.go |
| @@ -0,0 +1,168 @@ |
| +package live |
| + |
| +import ( |
| + "testing" |
| + "time" |
| +) |
| + |
| +// drain reads every event already queued on ch without blocking. |
| +func drain(ch <-chan Event) []Event { |
| + var out []Event |
| + for { |
| + select { |
| + case e, ok := <-ch: |
| + if !ok { |
| + return out |
| + } |
| + out = append(out, e) |
| + default: |
| + return out |
| + } |
| + } |
| +} |
| + |
| +// next takes the one event a test just caused, failing rather than |
| +// hanging when the hub said nothing. |
| +func next(t *testing.T, ch <-chan Event) Event { |
| + t.Helper() |
| + select { |
| + case e, ok := <-ch: |
| + if !ok { |
| + t.Fatal("the subscriber was dropped") |
| + } |
| + return e |
| + default: |
| + t.Fatal("no event") |
| + } |
| + return Event{} |
| +} |
| + |
| +// TestSeqIsAdvisoryAndOnlyReachesItsOwnDeck: a published seq goes to |
| +// everyone watching that deck and to nobody watching another one. |
| +func TestSeqIsAdvisoryAndOnlyReachesItsOwnDeck(t *testing.T) { |
| + h := New() |
| + defer h.Close() |
| + |
| + mine, stop := h.Subscribe("deck-a") |
| + defer stop() |
| + theirs, stopTheirs := h.Subscribe("deck-b") |
| + defer stopTheirs() |
| + |
| + h.PublishSeq("deck-a", 7) |
| + |
| + e := next(t, mine) |
| + if e.Seq != 7 || e.Presence != nil { |
| + t.Errorf("the subscriber heard %+v, want seq 7 and no presence", e) |
| + } |
| + if got := drain(theirs); len(got) != 0 { |
| + t.Errorf("a subscriber on another deck heard %+v", got) |
| + } |
| +} |
| + |
| +// TestPresenceIsAnnouncedAndExpires: posting presence publishes the |
| +// WHOLE set for that deck, and a person who stops heartbeating is |
| +// swept out and the remaining set republished. |
| +func TestPresenceIsAnnouncedAndExpires(t *testing.T) { |
| + h := New() |
| + defer h.Close() |
| + |
| + now := time.Date(2026, 9, 7, 12, 0, 0, 0, time.UTC) |
| + h.now = func() time.Time { return now } |
| + |
| + ch, stop := h.Subscribe("deck-a") |
| + defer stop() |
| + |
| + h.PublishPresence("deck-a", Presence{Client: "tab-1", Name: "Paul", Colour: "#111", Slide: "s1"}) |
| + e := next(t, ch) |
| + if e.Presence == nil { |
| + t.Fatalf("presence published a seq event: %+v", e) |
| + } |
| + if set := *e.Presence; len(set) != 1 || set[0].Client != "tab-1" || set[0].Name != "Paul" { |
| + t.Fatalf("the announced set is %+v, want one entry for tab-1", set) |
| + } |
| + |
| + // A second tab: the set, never a delta. |
| + h.PublishPresence("deck-a", Presence{Client: "tab-2", Name: "Sam", Colour: "#222"}) |
| + if set := *next(t, ch).Presence; len(set) != 2 { |
| + t.Fatalf("the announced set is %+v, want both tabs", set) |
| + } |
| + |
| + // tab-1 keeps heartbeating; tab-2 goes quiet. |
| + now = now.Add(30 * time.Second) |
| + h.PublishPresence("deck-a", Presence{Client: "tab-1", Name: "Paul", Colour: "#111", Slide: "s2"}) |
| + drain(ch) |
| + |
| + // Past two missed heartbeats the sweeper drops tab-2 and says so. |
| + now = now.Add(20 * time.Second) |
| + h.sweep() |
| + e = next(t, ch) |
| + if e.Presence == nil { |
| + t.Fatalf("the sweep published a seq event: %+v", e) |
| + } |
| + set := *e.Presence |
| + if len(set) != 1 || set[0].Client != "tab-1" { |
| + t.Fatalf("after the sweep the set is %+v, want tab-1 alone", set) |
| + } |
| + if got := h.Snapshot("deck-a"); len(got) != 1 || got[0].Slide != "s2" { |
| + t.Errorf("Snapshot is %+v, want tab-1 on s2", got) |
| + } |
| + |
| + // A sweep that changes nothing says nothing. |
| + h.sweep() |
| + if got := drain(ch); len(got) != 0 { |
| + t.Errorf("an idle sweep published %+v", got) |
| + } |
| + |
| + // Everyone gone: the deck is forgotten rather than kept as an |
| + // empty entry for the life of the process. |
| + now = now.Add(time.Minute) |
| + h.sweep() |
| + if set := *next(t, ch).Presence; len(set) != 0 { |
| + t.Errorf("the final sweep announced %+v, want an empty set", set) |
| + } |
| + if n := h.presenceDecks(); n != 0 { |
| + t.Errorf("the hub still holds presence for %d decks, want 0", n) |
| + } |
| +} |
| + |
| +// TestSubscriberThatFallsBehindIsDropped: presence is whole state, so |
| +// a reader that cannot keep up is closed and left to reconnect rather |
| +// than fed a set with holes in it. |
| +func TestSubscriberThatFallsBehindIsDropped(t *testing.T) { |
| + h := New() |
| + defer h.Close() |
| + |
| + ch, stop := h.Subscribe("deck-a") |
| + defer stop() |
| + for i := 0; i < subscriberBuffer+2; i++ { |
| + h.PublishSeq("deck-a", int64(i+1)) |
| + } |
| + for i := 0; i < subscriberBuffer; i++ { |
| + if _, ok := <-ch; !ok { |
| + t.Fatalf("the subscriber closed after %d events, want %d buffered first", i, subscriberBuffer) |
| + } |
| + } |
| + if _, ok := <-ch; ok { |
| + t.Fatal("a subscriber that fell behind is still open") |
| + } |
| + if n := h.deckCount(); n != 0 { |
| + t.Errorf("the dropped subscriber left %d decks behind", n) |
| + } |
| +} |
| + |
| +// TestUnsubscribeForgetsTheDeck: the subscriber map must not grow one |
| +// empty entry per deck ever opened. |
| +func TestUnsubscribeForgetsTheDeck(t *testing.T) { |
| + h := New() |
| + defer h.Close() |
| + |
| + _, stop := h.Subscribe("deck-a") |
| + if n := h.deckCount(); n != 1 { |
| + t.Fatalf("deckCount = %d after one subscribe, want 1", n) |
| + } |
| + stop() |
| + stop() // idempotent |
| + if n := h.deckCount(); n != 0 { |
| + t.Errorf("deckCount = %d after unsubscribing, want 0", n) |
| + } |
| +} |
diff --git a/internal/slides/locales/en.toml b/internal/slides/locales/en.toml| index 81ecb42..429e08c 100644 |
| --- a/internal/slides/locales/en.toml |
| +++ b/internal/slides/locales/en.toml |
| @@ -273,3 +273,13 @@ slides.remote.pair.error = "That code and key do not go together. Check both, or |
| # moves them. |
| slides.follow.offer = "Follow the presenter" |
| slides.follow.browse = "Browse on your own" |
| + |
| +# slides.live.* — the live-sync transport (live.go). These four are |
| +# read by a script rather than by a person, so each is written for the |
| +# person who WILL read it: whoever has the network tab open when |
| +# something has gone wrong. Plain sentences, no codes, and never a |
| +# reason the client cannot act on. |
| +slides.live.error.badsince = "That cursor isn't a number. Reload the page." |
| +slides.live.error.toolarge = "That change is too large to save in one piece." |
| +slides.live.error.invalid = "This change couldn't be applied. Reload the page and try again." |
| +slides.live.error.busy = "Somebody else saved at the same moment. Try again." |
diff --git a/internal/slides/migrations/0006_create_deck_snapshots.sql b/internal/slides/migrations/0006_create_deck_snapshots.sql| new file mode 100644 |
| index 0000000..521f382 |
| --- /dev/null |
| +++ b/internal/slides/migrations/0006_create_deck_snapshots.sql |
| @@ -0,0 +1,4 @@ |
| +CREATE TABLE `deck_snapshots` (`id` integer PRIMARY KEY AUTOINCREMENT,`deck_id` integer,`seq` integer,`bytes` blob,`created_at` datetime); |
| + |
| +CREATE INDEX `idx_deck_snapshots_deck_id` ON `deck_snapshots`(`deck_id`); |
| + |
diff --git a/internal/slides/migrations/schema.sql b/internal/slides/migrations/schema.sql| index 5a1e998..a65a53b 100644 |
| --- a/internal/slides/migrations/schema.sql |
| +++ b/internal/slides/migrations/schema.sql |
| @@ -5,6 +5,8 @@ CREATE TABLE `agents` (`id` integer PRIMARY KEY AUTOINCREMENT,`name` text,`opera |
| CREATE TABLE `deck_grants` (`id` integer PRIMARY KEY AUTOINCREMENT,`deck_id` integer,`user_id` integer,`actor` text,`created_at` datetime); |
| +CREATE TABLE `deck_snapshots` (`id` integer PRIMARY KEY AUTOINCREMENT,`deck_id` integer,`seq` integer,`bytes` blob,`created_at` datetime); |
| + |
| CREATE TABLE `deck_updates` (`id` integer PRIMARY KEY AUTOINCREMENT,`deck_id` integer,`seq` integer,`actor` text,`bytes` blob,`created_at` datetime); |
| CREATE TABLE `decks` (`id` integer PRIMARY KEY AUTOINCREMENT,`sqid` text,`title` text,`owner_id` integer,`rung` text,`theme` text,`seq` integer,`created_at` datetime,`updated_at` datetime); |
| @@ -23,6 +25,8 @@ CREATE INDEX `idx_deck_grants_deck_id` ON `deck_grants`(`deck_id`); |
| CREATE INDEX `idx_deck_grants_user_id` ON `deck_grants`(`user_id`); |
| +CREATE INDEX `idx_deck_snapshots_deck_id` ON `deck_snapshots`(`deck_id`); |
| + |
| CREATE UNIQUE INDEX `idx_deck_updates_deck_seq` ON `deck_updates`(`deck_id`,`seq`); |
| CREATE INDEX `idx_decks_owner_id` ON `decks`(`owner_id`); |
diff --git a/internal/slides/models.go b/internal/slides/models.go| index ace5687..8f5ae40 100644 |
| --- a/internal/slides/models.go |
| +++ b/internal/slides/models.go |
| @@ -11,6 +11,7 @@ var Models = []any{ |
| &User{}, |
| &Deck{}, |
| &DeckUpdate{}, |
| + &DeckSnapshot{}, |
| &DeckGrant{}, |
| &ShareLink{}, |
| &Presentation{}, |
| @@ -113,6 +114,42 @@ type DeckUpdate struct { |
| CreatedAt time.Time |
| } |
| +// DeckSnapshot is one deck's whole document at one seq, stored as a |
| +// single Yjs update (core.Handle.State) so a later open can start from |
| +// it instead of from the first row of the log. |
| +// |
| +// It is DERIVED DATA, and every decision here follows from that (Docs' |
| +// snapshots ruling, adopted). The log is what the deck is: delete every |
| +// row of this table and nothing is lost but time. So a snapshot is |
| +// written best-effort AFTER the update that crossed the boundary has |
| +// committed (snapshot.go's maybeSnapshot), a failure to write one is |
| +// logged and never fails the request, and a snapshot that cannot be |
| +// read is skipped in favour of replaying the plain log. |
| +// |
| +// Seq is the highest seq the bytes contain, and openDeck trusts it: it |
| +// replays the rows at seq > Seq and no others. Nothing here can check |
| +// that claim, so the ONE caller allowed to write a row is the one |
| +// holding a handle it knows the seq of. |
| +// |
| +// Bytes rather than a path: a deck's state is kilobytes, this is one |
| +// row per two hundred updates, and superseded rows are deleted as each |
| +// new one lands, so the table holds one row per deck in the steady |
| +// state. |
| +// |
| +// (deck_id, seq) is deliberately NOT unique. Two writers crossing the |
| +// boundary together produce equivalent rows at the same seq, and that |
| +// is an ordinary outcome rather than a collision to resolve. |
| +type DeckSnapshot struct { |
| + ID int64 |
| + DeckID int64 `gorm:"index"` |
| + Seq int64 |
| + Bytes []byte |
| + // CreatedAt is when the snapshot was taken, not when the deck last |
| + // moved: a snapshot is bookkeeping, and the deck's own UpdatedAt is |
| + // the column anything user-facing reads. |
| + CreatedAt time.Time |
| +} |
| + |
| // DeckGrant names one person who may open one deck — spec §5's |
| // `granted` rung. |
| // |
diff --git a/internal/slides/snapshot.go b/internal/slides/snapshot.go| new file mode 100644 |
| index 0000000..7ecda2b |
| --- /dev/null |
| +++ b/internal/slides/snapshot.go |
| @@ -0,0 +1,152 @@ |
| +// snapshot.go is the accelerator under openDeck: a deck's whole |
| +// document written down every so often, so opening it replays a short |
| +// tail instead of every update since it was made. |
| +// |
| +// The rule the whole file is built on, and the one to keep if anything |
| +// here is ever rewritten: THE LOG IS THE DECK. A snapshot is derived |
| +// data (models.go's DeckSnapshot, Docs' snapshots ruling). Every path |
| +// in this file is therefore allowed to fail without failing the request |
| +// that reached it, and openDeck falls back to the plain log when it |
| +// does. |
| +package slides |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "time" |
| + |
| + "gorm.io/gorm" |
| + |
| + "amadan.net/oficina/slides/internal/slides/core" |
| +) |
| + |
| +// snapshotCadence is how many updates may accumulate before the next |
| +// snapshot is taken: the request whose seq lands on a multiple of this |
| +// writes one (maybeSnapshot). |
| +// |
| +// 200 is the plan's number, and the tail it leaves — at most 199 |
| +// updates to replay — sits about twenty times clear of |
| +// core.MaxRetainedUpdates (4096), which is the hard wall a handle hits |
| +// when it has replayed too much to rebuild itself. That headroom is |
| +// conditional on snapshot writes actually succeeding; each failure is |
| +// logged at Error for exactly that reason. |
| +const snapshotCadence = 200 |
| + |
| +// latestSnapshot returns the newest usable snapshot for deckID, or nil |
| +// when there is none. |
| +// |
| +// Ordering is seq DESC, id DESC: two writers crossing the boundary |
| +// together may legitimately write equivalent rows at the same seq, so |
| +// the tie resolves to the last row written rather than being treated as |
| +// an error worth reporting. |
| +func (a *app) latestSnapshot(ctx context.Context, deckID int64) (*DeckSnapshot, error) { |
| + var row DeckSnapshot |
| + err := a.db.G.WithContext(ctx). |
| + Where("deck_id = ?", deckID). |
| + Order("seq DESC, id DESC"). |
| + Take(&row).Error |
| + if errors.Is(err, gorm.ErrRecordNotFound) { |
| + return nil, nil |
| + } |
| + if err != nil { |
| + return nil, fmt.Errorf("slides: newest snapshot for deck %d: %w", deckID, err) |
| + } |
| + return &row, nil |
| +} |
| + |
| +// replayPlan is what openDeck has to apply to reconstitute d: the |
| +// newest snapshot (or nil) and every update after it, in seq order. |
| +// |
| +// It is one function rather than two calls at the call site so that the |
| +// invariant tying them together — the tail starts exactly where the |
| +// snapshot ends — is written once and can be tested on its own |
| +// (snapshot_test.go). |
| +// |
| +// A snapshot that cannot be read is not an error here. The log alone |
| +// still answers the question, so the failure is returned to the caller |
| +// as "no snapshot" plus a log line, and the deck opens slowly rather |
| +// than not at all. |
| +func (a *app) replayPlan(ctx context.Context, d Deck) (*DeckSnapshot, []DeckUpdate, error) { |
| + snap, err := a.latestSnapshot(ctx, d.ID) |
| + if err != nil { |
| + a.logger.Error("slides: read a deck's newest snapshot, replaying the plain log instead", |
| + "deck", d.Sqid, "err", err) |
| + snap = nil |
| + } |
| + |
| + var since int64 |
| + if snap != nil { |
| + since = snap.Seq |
| + } |
| + var updates []DeckUpdate |
| + if err := a.db.G.WithContext(ctx). |
| + Where("deck_id = ? AND seq > ?", d.ID, since). |
| + Order("seq").Find(&updates).Error; err != nil { |
| + return nil, nil, fmt.Errorf("slides: read deck %s log: %w", d.Sqid, err) |
| + } |
| + return snap, updates, nil |
| +} |
| + |
| +// maybeSnapshot writes a snapshot when seq lands on the cadence |
| +// boundary, and does nothing at all otherwise. |
| +// |
| +// Two things it is the caller's job to have true, because nothing here |
| +// can check either: |
| +// |
| +// - h holds the document THROUGH seq and no further. Every caller |
| +// applies its update to h and commits it before calling this, so |
| +// the number on the row is the number in the bytes. |
| +// - The transaction that wrote seq has COMMITTED. A snapshot taken |
| +// from a rolled-back state would be a copy of a deck that never |
| +// existed, and openDeck would then skip the real rows it claims to |
| +// cover. |
| +// |
| +// Every failure is logged and swallowed. This runs after the update the |
| +// person actually made is safely on disk; failing their request because |
| +// the accelerator could not be written would be trading the thing that |
| +// matters for the thing that does not. |
| +func (a *app) maybeSnapshot(ctx context.Context, d Deck, h *core.Handle, seq int64) { |
| + if seq <= 0 || seq%snapshotCadence != 0 { |
| + return |
| + } |
| + |
| + state, err := h.State(ctx) |
| + if err != nil { |
| + a.logger.Error("slides: encode a deck snapshot", "deck", d.Sqid, "seq", seq, "err", err) |
| + return |
| + } |
| + |
| + row := &DeckSnapshot{ |
| + DeckID: d.ID, |
| + Seq: seq, |
| + Bytes: state, |
| + CreatedAt: time.Now().UTC(), |
| + } |
| + err = a.db.G.WithContext(ctx).Transaction(func(tx *gorm.DB) error { |
| + if err := tx.Create(row).Error; err != nil { |
| + return fmt.Errorf("slides: persist a snapshot for deck %s at seq %d: %w", d.Sqid, seq, err) |
| + } |
| + // Superseded snapshots are deleted rather than kept: they are |
| + // reproducible from a log that keeps every row, so this deletes |
| + // no history, and retaining them would cost one whole copy of |
| + // the deck per two hundred updates — the difference between an |
| + // accelerator and a second, fatter archive. |
| + // |
| + // "Superseded" is (seq, id) ordered exactly as latestSnapshot |
| + // reads, so what survives is always what latestSnapshot would |
| + // return, and a row at a HIGHER seq — written by another |
| + // request that got further — is never touched. "Delete |
| + // everything that is not mine" would take that one out and |
| + // quietly put the deck back to replaying updates it had |
| + // already been spared. |
| + if err := tx.Where("deck_id = ? AND (seq < ? OR (seq = ? AND id < ?))", |
| + d.ID, row.Seq, row.Seq, row.ID).Delete(&DeckSnapshot{}).Error; err != nil { |
| + return fmt.Errorf("slides: delete superseded snapshots for deck %s at seq %d: %w", d.Sqid, seq, err) |
| + } |
| + return nil |
| + }) |
| + if err != nil { |
| + a.logger.Error("slides: write a deck snapshot", "deck", d.Sqid, "seq", seq, "err", err) |
| + } |
| +} |
diff --git a/internal/slides/snapshot_test.go b/internal/slides/snapshot_test.go| new file mode 100644 |
| index 0000000..c25f4c0 |
| --- /dev/null |
| +++ b/internal/slides/snapshot_test.go |
| @@ -0,0 +1,125 @@ |
| +package slides |
| + |
| +import ( |
| + "context" |
| + "fmt" |
| + "testing" |
| +) |
| + |
| +// TestSnapshotWrittenAtBoundaryAndUsedOnOpen: a deck that crosses the |
| +// cadence boundary gets exactly one snapshot, openDeck replays only the |
| +// tail after it, and the document it produces is the same document the |
| +// whole log produces. |
| +// |
| +// The last clause is the one that matters. A snapshot is an accelerator |
| +// and derived data (models.go's DeckSnapshot); the moment it is not |
| +// byte-for-byte the same deck, it is a second, wrong copy of the truth. |
| +func TestSnapshotWrittenAtBoundaryAndUsedOnOpen(t *testing.T) { |
| + d := testDB(t) |
| + a, _, err := newApp(d, Config{Env: EnvDevelopment, Logger: quietLogger()}) |
| + if err != nil { |
| + t.Fatalf("newApp: %v", err) |
| + } |
| + t.Cleanup(func() { a.Close() }) |
| + ctx := context.Background() |
| + |
| + owner, _, err := ensureUser(ctx, d.G, "paul@example.test") |
| + if err != nil { |
| + t.Fatalf("ensureUser: %v", err) |
| + } |
| + actor := actorFor(owner) |
| + |
| + row, err := a.createDeckRow(ctx, owner, "Long deck", defaultTheme) |
| + if err != nil { |
| + t.Fatalf("createDeckRow: %v", err) |
| + } |
| + opening, err := openingOps("Long deck", defaultTheme) |
| + if err != nil { |
| + t.Fatalf("openingOps: %v", err) |
| + } |
| + if err := a.editDeck(ctx, &row, actor, opening, nil); err != nil { |
| + t.Fatalf("editDeck: %v", err) |
| + } |
| + |
| + // One handle for the rest, so this is 200 applies rather than 200 |
| + // full replays — the test is about the boundary, not about how |
| + // slowly a deck can be built. |
| + h, _, err := a.openDeck(ctx, row) |
| + if err != nil { |
| + t.Fatalf("openDeck: %v", err) |
| + } |
| + defer h.Close(ctx) |
| + for row.Seq < snapshotCadence+1 { |
| + batch, err := opsJSON([]op{{"op": "set_meta", "title": fmt.Sprintf("Long deck %d", row.Seq)}}) |
| + if err != nil { |
| + t.Fatalf("opsJSON: %v", err) |
| + } |
| + if _, err := a.applyOps(ctx, h, &row, actor, batch, nil); err != nil { |
| + t.Fatalf("applyOps at seq %d: %v", row.Seq, err) |
| + } |
| + } |
| + if row.Seq != snapshotCadence+1 { |
| + t.Fatalf("the deck is at seq %d, want %d", row.Seq, snapshotCadence+1) |
| + } |
| + |
| + var snapshots []DeckSnapshot |
| + if err := d.G.WithContext(ctx).Where("deck_id = ?", row.ID).Order("seq").Find(&snapshots).Error; err != nil { |
| + t.Fatalf("read snapshots: %v", err) |
| + } |
| + if len(snapshots) != 1 { |
| + t.Fatalf("%d snapshots after %d updates, want exactly 1", len(snapshots), row.Seq) |
| + } |
| + if snapshots[0].Seq != snapshotCadence { |
| + t.Errorf("the snapshot is at seq %d, want %d", snapshots[0].Seq, snapshotCadence) |
| + } |
| + if len(snapshots[0].Bytes) == 0 { |
| + t.Error("the snapshot carries no bytes") |
| + } |
| + |
| + // openDeck replays the tail and nothing before it. |
| + snap, tail, err := a.replayPlan(ctx, row) |
| + if err != nil { |
| + t.Fatalf("replayPlan: %v", err) |
| + } |
| + if snap == nil || snap.Seq != snapshotCadence { |
| + t.Fatalf("replayPlan chose %+v, want the snapshot at seq %d", snap, snapshotCadence) |
| + } |
| + if len(tail) != 1 || tail[0].Seq != row.Seq { |
| + t.Fatalf("replayPlan replays %d updates, want the one at seq %d", len(tail), row.Seq) |
| + } |
| + |
| + // And it is the same deck either way. |
| + want, err := h.JSON(ctx) |
| + if err != nil { |
| + t.Fatalf("read the live view: %v", err) |
| + } |
| + fromSnapshot := viewJSON(t, a, row) |
| + if string(fromSnapshot) != string(want) { |
| + t.Fatalf("opening from the snapshot gives a different deck:\n got %s\nwant %s", fromSnapshot, want) |
| + } |
| + |
| + // Deleting the snapshot loses nothing: the log alone still is the |
| + // deck, which is what makes a snapshot safe to throw away. |
| + if err := d.G.WithContext(ctx).Where("deck_id = ?", row.ID).Delete(&DeckSnapshot{}).Error; err != nil { |
| + t.Fatalf("delete snapshots: %v", err) |
| + } |
| + if fromLog := viewJSON(t, a, row); string(fromLog) != string(want) { |
| + t.Fatalf("opening from the plain log gives a different deck:\n got %s\nwant %s", fromLog, want) |
| + } |
| +} |
| + |
| +// viewJSON opens d the way a request does and returns the core's view. |
| +func viewJSON(t *testing.T, a *app, d Deck) []byte { |
| + t.Helper() |
| + ctx := context.Background() |
| + h, _, err := a.openDeck(ctx, d) |
| + if err != nil { |
| + t.Fatalf("openDeck: %v", err) |
| + } |
| + defer h.Close(ctx) |
| + raw, err := h.JSON(ctx) |
| + if err != nil { |
| + t.Fatalf("read the view: %v", err) |
| + } |
| + return raw |
| +} |
diff --git a/internal/slidestest/live_test.go b/internal/slidestest/live_test.go| new file mode 100644 |
| index 0000000..67800f7 |
| --- /dev/null |
| +++ b/internal/slidestest/live_test.go |
| @@ -0,0 +1,511 @@ |
| +package slidestest |
| + |
| +import ( |
| + "bufio" |
| + "bytes" |
| + "context" |
| + "encoding/binary" |
| + "net/http" |
| + "net/http/httptest" |
| + "net/url" |
| + "strconv" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "amadan.net/oficina/slides/internal/slides/core" |
| +) |
| + |
| +// --- helpers --- |
| + |
| +// postBody makes a POST with a body of this test's choosing — the |
| +// binary update endpoint and the JSON presence endpoint both take one, |
| +// and neither is a form. |
| +// |
| +// It sends the same same-origin evidence `post` does: these are |
| +// browser-made requests, inside csrf.Protect like every other write. |
| +func postBody(t *testing.T, a *testApp, target, contentType string, body []byte, cookies []*http.Cookie) *httptest.ResponseRecorder { |
| + t.Helper() |
| + req := httptest.NewRequest(http.MethodPost, target, bytes.NewReader(body)) |
| + req.Header.Set("Content-Type", contentType) |
| + req.Header.Set("Origin", testOrigin) |
| + req.Header.Set("Sec-Fetch-Site", "same-origin") |
| + for _, c := range cookies { |
| + req.AddCookie(c) |
| + } |
| + rec := httptest.NewRecorder() |
| + a.ServeHTTP(rec, req) |
| + return rec |
| +} |
| + |
| +// frames splits an updates response into the updates it carries: a |
| +// 4-byte big-endian length then that many bytes, repeated. A body that |
| +// does not divide cleanly is a failure rather than a short read — the |
| +// framing is the protocol, and a client that guessed would be a client |
| +// that silently loses the last update. |
| +func frames(t *testing.T, body []byte) [][]byte { |
| + t.Helper() |
| + var out [][]byte |
| + for len(body) > 0 { |
| + if len(body) < 4 { |
| + t.Fatalf("%d trailing bytes with no length prefix", len(body)) |
| + } |
| + n := binary.BigEndian.Uint32(body[:4]) |
| + body = body[4:] |
| + if uint32(len(body)) < n { |
| + t.Fatalf("a frame claims %d bytes and %d remain", n, len(body)) |
| + } |
| + out = append(out, body[:n]) |
| + body = body[n:] |
| + } |
| + return out |
| +} |
| + |
| +// getUpdates fetches one cursor's worth of the log and returns the |
| +// frames plus the advisory seq the response carried. |
| +func getUpdates(t *testing.T, a *testApp, path string, since int64, cookies []*http.Cookie) ([][]byte, int64) { |
| + t.Helper() |
| + target := path + "/updates?since=" + strconv.FormatInt(since, 10) |
| + rec := mustGet(t, a, target, http.StatusOK, cookies) |
| + if ct := rec.Header().Get("Content-Type"); ct != "application/octet-stream" { |
| + t.Fatalf("GET %s: Content-Type = %q, want application/octet-stream", target, ct) |
| + } |
| + seq, err := strconv.ParseInt(rec.Header().Get("Slides-Seq"), 10, 64) |
| + if err != nil { |
| + t.Fatalf("GET %s: Slides-Seq = %q, not a number", target, rec.Header().Get("Slides-Seq")) |
| + } |
| + return frames(t, rec.Body.Bytes()), seq |
| +} |
| + |
| +// mirror is a browser's copy of a deck: a core document of its own, |
| +// caught up to the server's log, that produces the update bytes a real |
| +// editor would post. It is the only honest way to test this wire — |
| +// bytes made up by hand are not Yjs updates and prove nothing about the |
| +// path they travel. |
| +type mirror struct { |
| + rt *core.Runtime |
| + h *core.Handle |
| +} |
| + |
| +func newMirror(t *testing.T, sqid string, log [][]byte) *mirror { |
| + t.Helper() |
| + ctx := context.Background() |
| + rt, err := core.New(ctx) |
| + if err != nil { |
| + t.Fatalf("core.New: %v", err) |
| + } |
| + // Client id 2: the server writes as 1 (decks.go's serverClientID), |
| + // and two writers sharing one client id is the one thing a CRDT |
| + // cannot forgive. |
| + h, err := rt.NewDoc(ctx, sqid, 2) |
| + if err != nil { |
| + rt.Close(ctx) |
| + t.Fatalf("core.NewDoc: %v", err) |
| + } |
| + m := &mirror{rt: rt, h: h} |
| + t.Cleanup(func() { |
| + h.Close(ctx) |
| + rt.Close(ctx) |
| + }) |
| + m.apply(t, log) |
| + return m |
| +} |
| + |
| +func (m *mirror) apply(t *testing.T, log [][]byte) { |
| + t.Helper() |
| + for i, u := range log { |
| + if err := m.h.Apply(context.Background(), u); err != nil { |
| + t.Fatalf("applying frame %d to the mirror: %v", i, err) |
| + } |
| + } |
| +} |
| + |
| +// ops runs a batch against the mirror and returns the update it |
| +// produced — exactly what the browser's own ops module will POST. |
| +func (m *mirror) ops(t *testing.T, batch string) []byte { |
| + t.Helper() |
| + update, err := m.h.Ops(context.Background(), []byte(batch)) |
| + if err != nil { |
| + t.Fatalf("mirror ops %s: %v", batch, err) |
| + } |
| + return update |
| +} |
| + |
| +// postUpdate posts one update and insists on the 204 and the seq. |
| +func postUpdate(t *testing.T, a *testApp, path string, update []byte, cookies []*http.Cookie) int64 { |
| + t.Helper() |
| + rec := postBody(t, a, path+"/updates", "application/octet-stream", update, cookies) |
| + if rec.Code != http.StatusNoContent { |
| + t.Fatalf("POST %s/updates: status %d, want 204\n%s", path, rec.Code, rec.Body) |
| + } |
| + seq, err := strconv.ParseInt(rec.Header().Get("Slides-Seq"), 10, 64) |
| + if err != nil { |
| + t.Fatalf("POST %s/updates: Slides-Seq = %q, not a number", path, rec.Header().Get("Slides-Seq")) |
| + } |
| + return seq |
| +} |
| + |
| +// scan opens the deck's SSE stream against a real server and returns a |
| +// reader over it. The request's context is the timeout, so a stream |
| +// that never speaks fails the test rather than hanging it. |
| +func scan(t *testing.T, a *testApp, path string, cookies []*http.Cookie) *bufio.Reader { |
| + t.Helper() |
| + srv := httptest.NewServer(a) |
| + t.Cleanup(srv.Close) |
| + |
| + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| + t.Cleanup(cancel) |
| + req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+path+"/stream", nil) |
| + if err != nil { |
| + t.Fatalf("build the stream request: %v", err) |
| + } |
| + for _, c := range cookies { |
| + req.AddCookie(c) |
| + } |
| + resp, err := http.DefaultClient.Do(req) |
| + if err != nil { |
| + t.Fatalf("open the stream: %v", err) |
| + } |
| + t.Cleanup(func() { resp.Body.Close() }) |
| + if resp.StatusCode != http.StatusOK { |
| + t.Fatalf("GET %s/stream: status %d, want 200", path, resp.StatusCode) |
| + } |
| + if got := resp.Header.Get("Content-Type"); got != "text/event-stream" { |
| + t.Fatalf("the stream's Content-Type is %q, want text/event-stream", got) |
| + } |
| + return bufio.NewReader(resp.Body) |
| +} |
| + |
| +// nextEvent reads events until one of the named kind arrives, |
| +// returning its data line. Pings and the other kind are skipped. |
| +func nextEvent(t *testing.T, r *bufio.Reader, want string) string { |
| + t.Helper() |
| + var kind string |
| + for { |
| + line, err := r.ReadString('\n') |
| + if err != nil { |
| + t.Fatalf("reading the stream for a %q event: %v", want, err) |
| + } |
| + line = strings.TrimRight(line, "\r\n") |
| + switch { |
| + case strings.HasPrefix(line, "event: "): |
| + kind = strings.TrimPrefix(line, "event: ") |
| + case strings.HasPrefix(line, "data: "): |
| + if kind == want { |
| + return strings.TrimPrefix(line, "data: ") |
| + } |
| + } |
| + } |
| +} |
| + |
| +// scalar reads one value out of the app's own database, for the facts |
| +// that have no screen: which actor a row recorded, which id a person |
| +// got. |
| +func scalar(t *testing.T, a *testApp, query string, args ...any) string { |
| + t.Helper() |
| + var out string |
| + if err := a.db.G.Raw(query, args...).Scan(&out).Error; err != nil { |
| + t.Fatalf("query %q: %v", query, err) |
| + } |
| + return out |
| +} |
| + |
| +// sqidOf turns the /d/{sqid} path createDeck returns into the sqid. |
| +func sqidOf(path string) string { return strings.TrimPrefix(path, "/d/") } |
| + |
| +// --- the wire --- |
| + |
| +// TestUpdatesRoundTripThroughTheWire: a browser's own update, made by |
| +// the same core the server runs, posted, read back framed, and visible |
| +// in the deck the server renders. |
| +// |
| +// The last step is the point. A test that posted bytes and read the |
| +// same bytes back would pass over a table that stored them and a |
| +// document that never saw them. |
| +func TestUpdatesRoundTripThroughTheWire(t *testing.T) { |
| + a := newApp(t) |
| + cookies := signedIn(t, a, allowedAddress) |
| + path := createDeck(t, a, cookies, "Wire") |
| + |
| + // A new deck is one update: its opening batch (decks.go). |
| + log, seq := getUpdates(t, a, path, 0, cookies) |
| + if len(log) != 1 { |
| + t.Fatalf("a new deck's log has %d frames, want 1", len(log)) |
| + } |
| + if seq != 1 { |
| + t.Fatalf("Slides-Seq on a new deck is %d, want 1", seq) |
| + } |
| + |
| + m := newMirror(t, sqidOf(path), log) |
| + update := m.ops(t, `[{"op":"add_slide","id":"wiretestslide","layout":"flow","template":"title"}]`) |
| + |
| + if got := postUpdate(t, a, path, update, cookies); got != 2 { |
| + t.Fatalf("POST answered Slides-Seq %d, want 2", got) |
| + } |
| + |
| + // Read back from the cursor the client held: exactly the one |
| + // update it did not already have, byte for byte. |
| + tail, seq := getUpdates(t, a, path, 1, cookies) |
| + if len(tail) != 1 { |
| + t.Fatalf("since=1 returned %d frames, want 1", len(tail)) |
| + } |
| + if !bytes.Equal(tail[0], update) { |
| + t.Errorf("the frame read back is not the update posted:\n got %x\nwant %x", tail[0], update) |
| + } |
| + if seq != 2 { |
| + t.Errorf("Slides-Seq is %d, want 2", seq) |
| + } |
| + |
| + // And the server's own document really has the slide. |
| + rec := mustGet(t, a, path+"/slides/wiretestslide.html", http.StatusOK, cookies) |
| + if !strings.Contains(rec.Body.String(), "<section slides-slide") { |
| + t.Errorf("the slide the update added does not render:\n%s", rec.Body) |
| + } |
| +} |
| + |
| +// TestUpdatesSinceCursorIsExclusive: ?since=N means "everything after |
| +// N", never "everything from N". Off by one here and every client |
| +// re-applies the update it already holds on every reconnect. |
| +func TestUpdatesSinceCursorIsExclusive(t *testing.T) { |
| + a := newApp(t) |
| + cookies := signedIn(t, a, allowedAddress) |
| + path := createDeck(t, a, cookies, "Cursors") |
| + |
| + log, _ := getUpdates(t, a, path, 0, cookies) |
| + m := newMirror(t, sqidOf(path), log) |
| + postUpdate(t, a, path, m.ops(t, `[{"op":"set_meta","title":"Second"}]`), cookies) |
| + postUpdate(t, a, path, m.ops(t, `[{"op":"set_meta","title":"Third"}]`), cookies) |
| + |
| + for _, tc := range []struct { |
| + since int64 |
| + frames int |
| + seq int64 |
| + }{ |
| + {since: 0, frames: 3, seq: 3}, |
| + {since: 1, frames: 2, seq: 3}, |
| + {since: 2, frames: 1, seq: 3}, |
| + // Nothing after the last one, and the header says where the |
| + // client already is rather than dropping to zero. |
| + {since: 3, frames: 0, seq: 3}, |
| + // A cursor past the end is a client that is ahead of this |
| + // instance, not an error: it gets nothing and keeps its place. |
| + {since: 99, frames: 0, seq: 99}, |
| + } { |
| + got, seq := getUpdates(t, a, path, tc.since, cookies) |
| + if len(got) != tc.frames { |
| + t.Errorf("since=%d returned %d frames, want %d", tc.since, len(got), tc.frames) |
| + } |
| + if seq != tc.seq { |
| + t.Errorf("since=%d answered Slides-Seq %d, want %d", tc.since, seq, tc.seq) |
| + } |
| + } |
| + |
| + // A cursor that is not a number is a client bug, and it is told so |
| + // rather than being handed the whole log as if it had asked. |
| + mustGet(t, a, path+"/updates?since=banana", http.StatusBadRequest, cookies) |
| + mustGet(t, a, path+"/updates?since=-1", http.StatusBadRequest, cookies) |
| +} |
| + |
| +// TestStreamAnnouncesSeqOnConnectAndAfterPost: the stream says where |
| +// the deck is the moment it opens — which is what heals a tab whose |
| +// cursor fell behind — and says so again every time an update lands. |
| +func TestStreamAnnouncesSeqOnConnectAndAfterPost(t *testing.T) { |
| + a := newApp(t) |
| + cookies := signedIn(t, a, allowedAddress) |
| + path := createDeck(t, a, cookies, "Stream") |
| + |
| + log, _ := getUpdates(t, a, path, 0, cookies) |
| + m := newMirror(t, sqidOf(path), log) |
| + |
| + events := scan(t, a, path, cookies) |
| + if data := nextEvent(t, events, "update"); data != "1" { |
| + t.Errorf("the connect event is %q, want the deck's current seq 1", data) |
| + } |
| + // Nobody has posted presence, so the connect set is empty — an |
| + // event all the same, so a tab knows it is alone rather than |
| + // waiting to find out. |
| + if data := nextEvent(t, events, "presence"); data != "[]" { |
| + t.Errorf("the connect presence is %q, want an empty set", data) |
| + } |
| + |
| + postUpdate(t, a, path, m.ops(t, `[{"op":"set_meta","title":"Moved"}]`), cookies) |
| + if data := nextEvent(t, events, "update"); data != "2" { |
| + t.Errorf("after a POST the stream said %q, want 2", data) |
| + } |
| + |
| + // A write through any other path says so too: the stream is the |
| + // deck's, not the update endpoint's. |
| + rec := post(t, a, path+"/settings", url.Values{ |
| + "title": {"Renamed"}, "theme": {"ink"}, "rung": {"private"}, |
| + }, cookies...) |
| + if rec.Code != http.StatusSeeOther { |
| + t.Fatalf("POST %s/settings: status %d, want 303", path, rec.Code) |
| + } |
| + if data := nextEvent(t, events, "update"); data != "3" { |
| + t.Errorf("after a settings save the stream said %q, want 3", data) |
| + } |
| +} |
| + |
| +// TestPresencePostReachesTheStream: a heartbeat becomes the whole set, |
| +// stamped with the name and colour of the person whose SESSION it came |
| +// from rather than whatever the body claimed. |
| +func TestPresencePostReachesTheStream(t *testing.T) { |
| + a := newSharedApp(t) |
| + cookies := signedIn(t, a, allowedAddress) |
| + path := createDeck(t, a, cookies, "Presence") |
| + |
| + events := scan(t, a, path, cookies) |
| + nextEvent(t, events, "presence") |
| + |
| + body := `{"client":"tab-1","slide":"s1","block":"b1","name":"Somebody Else","colour":"#ff0000"}` |
| + rec := postBody(t, a, path+"/presence", "application/json", []byte(body), cookies) |
| + if rec.Code != http.StatusNoContent { |
| + t.Fatalf("POST %s/presence: status %d, want 204\n%s", path, rec.Code, rec.Body) |
| + } |
| + |
| + data := nextEvent(t, events, "presence") |
| + for _, want := range []string{`"client":"tab-1"`, `"slide":"s1"`, `"block":"b1"`, allowedAddress} { |
| + if !strings.Contains(data, want) { |
| + t.Errorf("the presence event %q does not carry %q", data, want) |
| + } |
| + } |
| + colour := scalar(t, a, "SELECT colour FROM users WHERE email = ?", allowedAddress) |
| + if colour == "" || !strings.Contains(data, `"colour":"`+colour+`"`) { |
| + t.Errorf("the presence event %q does not carry the person's own colour %q", data, colour) |
| + } |
| + if strings.Contains(data, "Somebody Else") || strings.Contains(data, "#ff0000") { |
| + t.Errorf("the body's own name or colour was trusted:\n%s", data) |
| + } |
| + |
| + // A heartbeat with no client to key on is a client bug, not a |
| + // person: refused rather than filed under the empty string. |
| + rec = postBody(t, a, path+"/presence", "application/json", []byte(`{"slide":"s1"}`), cookies) |
| + if rec.Code != http.StatusUnprocessableEntity { |
| + t.Errorf("POST presence with no client: status %d, want 422", rec.Code) |
| + } |
| +} |
| + |
| +// TestUpdatesRequireMembershipAndRecordActor: every live-sync route is |
| +// behind the same visibility seam every other deck route is, answering |
| +// 404 rather than 403 — and every write it accepts says who made it. |
| +func TestUpdatesRequireMembershipAndRecordActor(t *testing.T) { |
| + a := newSharedApp(t) |
| + mine := signedIn(t, a, allowedAddress) |
| + theirs := signedIn(t, a, otherAddress) |
| + path := createDeck(t, a, mine, "Members only") |
| + |
| + log, _ := getUpdates(t, a, path, 0, mine) |
| + m := newMirror(t, sqidOf(path), log) |
| + update := m.ops(t, `[{"op":"set_meta","title":"Mine"}]`) |
| + |
| + // A stranger cannot read the log, write to it, watch it, or say |
| + // they are in the deck. |
| + mustGet(t, a, path+"/updates?since=0", http.StatusNotFound, theirs) |
| + mustGet(t, a, path+"/stream", http.StatusNotFound, theirs) |
| + mustGet(t, a, path+"/slides/whatever.html", http.StatusNotFound, theirs) |
| + if rec := postBody(t, a, path+"/updates", "application/octet-stream", update, theirs); rec.Code != http.StatusNotFound { |
| + t.Errorf("a stranger's POST /updates: status %d, want 404", rec.Code) |
| + } |
| + if rec := postBody(t, a, path+"/presence", "application/json", []byte(`{"client":"x"}`), theirs); rec.Code != http.StatusNotFound { |
| + t.Errorf("a stranger's POST /presence: status %d, want 404", rec.Code) |
| + } |
| + // Signed out is a redirect to sign in, never the log. |
| + mustGet(t, a, path+"/updates?since=0", http.StatusSeeOther, nil) |
| + |
| + // The owner's own write records them. |
| + postUpdate(t, a, path, update, mine) |
| + id := scalar(t, a, "SELECT id FROM users WHERE email = ?", allowedAddress) |
| + actor := scalar(t, a, |
| + "SELECT actor FROM deck_updates WHERE deck_id = (SELECT id FROM decks WHERE sqid = ?) ORDER BY seq DESC LIMIT 1", |
| + sqidOf(path)) |
| + if want := "person:" + id; actor != want { |
| + t.Errorf("the update recorded actor %q, want %q", actor, want) |
| + } |
| + |
| + // A team member can edit: the live routes are visibleDecks' whole |
| + // audience, not the owner alone. |
| + if rec := post(t, a, path+"/settings", url.Values{ |
| + "title": {"Members only"}, "theme": {"ink"}, "rung": {"team"}, |
| + }, mine...); rec.Code != http.StatusSeeOther { |
| + t.Fatalf("raising the rung: status %d, want 303", rec.Code) |
| + } |
| + shared, _ := getUpdates(t, a, path, 0, theirs) |
| + if len(shared) == 0 { |
| + t.Fatal("a team member cannot read the deck's log") |
| + } |
| + m.apply(t, shared[len(shared)-1:]) |
| + postUpdate(t, a, path, m.ops(t, `[{"op":"set_meta","title":"Ours"}]`), theirs) |
| + theirID := scalar(t, a, "SELECT id FROM users WHERE email = ?", otherAddress) |
| + actor = scalar(t, a, |
| + "SELECT actor FROM deck_updates WHERE deck_id = (SELECT id FROM decks WHERE sqid = ?) ORDER BY seq DESC LIMIT 1", |
| + sqidOf(path)) |
| + if want := "person:" + theirID; actor != want { |
| + t.Errorf("the team member's update recorded actor %q, want %q", actor, want) |
| + } |
| +} |
| + |
| +// TestRefusedUpdateIs422: bytes the core will not have are the client's |
| +// problem, and the answer says so without pretending anything was |
| +// stored. |
| +func TestRefusedUpdateIs422(t *testing.T) { |
| + a := newApp(t) |
| + cookies := signedIn(t, a, allowedAddress) |
| + path := createDeck(t, a, cookies, "Refusals") |
| + |
| + rec := postBody(t, a, path+"/updates", "application/octet-stream", |
| + []byte("this is not a Yjs update"), cookies) |
| + if rec.Code != http.StatusUnprocessableEntity { |
| + t.Fatalf("POST garbage: status %d, want 422\n%s", rec.Code, rec.Body) |
| + } |
| + |
| + // Nothing was written: the log is still the one opening update. |
| + log, seq := getUpdates(t, a, path, 0, cookies) |
| + if len(log) != 1 || seq != 1 { |
| + t.Errorf("after a refusal the log has %d frames at seq %d, want 1 at 1", len(log), seq) |
| + } |
| + |
| + // And an update larger than the cap is refused before it is read |
| + // into memory at all. |
| + rec = postBody(t, a, path+"/updates", "application/octet-stream", |
| + make([]byte, maxUpdateBytesForTest+1), cookies) |
| + if rec.Code != http.StatusRequestEntityTooLarge { |
| + t.Errorf("POST an oversized update: status %d, want 413", rec.Code) |
| + } |
| +} |
| + |
| +// maxUpdateBytesForTest is the app's own cap, restated here rather than |
| +// exported: the number is part of the wire contract (the plan's "limit |
| +// 1 MiB"), so a test that read it from the app could never notice it |
| +// changing. |
| +const maxUpdateBytesForTest = 1 << 20 |
| + |
| +// TestSlideHTMLTwinRendersOneSection: the editor's stage and strip |
| +// fetch slide HTML with a session cookie rather than a bearer token, |
| +// and get the same section the API serves — the section alone, never a |
| +// page. |
| +func TestSlideHTMLTwinRendersOneSection(t *testing.T) { |
| + a := newApp(t) |
| + cookies := signedIn(t, a, allowedAddress) |
| + path := createDeck(t, a, cookies, "Stage") |
| + |
| + log, _ := getUpdates(t, a, path, 0, cookies) |
| + m := newMirror(t, sqidOf(path), log) |
| + postUpdate(t, a, path, m.ops(t, `[{"op":"add_slide","id":"stageslide","layout":"flow","template":"title"}]`), cookies) |
| + |
| + for _, vp := range []string{"", "desktop", "phone", "thumb"} { |
| + target := path + "/slides/stageslide.html" |
| + if vp != "" { |
| + target += "?vp=" + vp |
| + } |
| + body := mustGet(t, a, target, http.StatusOK, cookies).Body.String() |
| + if !strings.HasPrefix(strings.TrimSpace(body), "<section slides-slide") { |
| + t.Errorf("vp=%q is not a bare slide section:\n%s", vp, body) |
| + } |
| + if strings.Contains(body, "<html") { |
| + t.Errorf("vp=%q carries a whole page:\n%s", vp, body) |
| + } |
| + } |
| + |
| + mustGet(t, a, path+"/slides/nosuchslide.html", http.StatusNotFound, cookies) |
| + mustGet(t, a, path+"/slides/stageslide", http.StatusNotFound, cookies) |
| +} |