feat: presença por sessão, eventos SSE em tempo real e autosave

- presence.go: hub SSE + registro de presença (TTL 30s por heartbeat)
- GET /api/events (EventSource, auth cookie/Bearer) transmite saved/presence;
  POST /api/presence publica {session, notes[]}; create/update/restore/delete
  publicam savedEvent com a session de origem (eco ignorado pelo autor)
- PUT/POST/restore aceitam campo opcional session
- UI: id de sessão em sessionStorage; selo 👥 na aba, na lista de notas e no
  topo do editor quando outra sessão tem a mesma nota aberta; aba LIMPA
  recarrega sozinha ao receber saved; aba SUJA abre o diálogo de conflito
  automaticamente (conflictPending evita diálogos empilhados)
- Autosave: debounce de 1,5s em qualquer alteração (rascunho vira nota),
  mensagem de revisão 'autosave', status 'salvando…', toggle nas Preferências
- Testes: SSE hello/presence/saved + presença 400/401 via httptest
This commit is contained in:
lgcosta committed 2026-11-04 13:35:33 -03:00
1 parent 3b91c1a509
commit a55591d208
9 files changed
+543 -15

No files matched your search

+77
View File
@@ -1,6 +1,7 @@
package api_test
import (
"bufio"
"bytes"
"encoding/json"
"io"
@@ -10,6 +11,7 @@ import (
"strings"
"testing"
"testing/fstest"
"time"
"github.com/lgcosta/notas-online/internal/api"
"github.com/lgcosta/notas-online/internal/store"
@@ -377,6 +379,81 @@ func TestSaveConflictViaAPI(t *testing.T) {
resp.Body.Close()
}
func TestPresenceAndSSEEvents(t *testing.T) {
srv := newTestServer(t)
call := func(method, path, body string) *http.Response {
return do(t, srv, method, path, body, testToken, "")
}
// abre o stream SSE
req, err := http.NewRequest("GET", srv.URL+"/api/events?session=s1", nil)
if err != nil {
t.Fatal(err)
}
req.Header.Set("Authorization", "Bearer "+testToken)
streamResp, err := srv.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer streamResp.Body.Close()
if ct := streamResp.Header.Get("Content-Type"); ct != "text/event-stream" {
t.Fatalf("content-type SSE: %q", ct)
}
sc := bufio.NewScanner(streamResp.Body)
readEvent := func(contains string) string {
t.Helper()
deadline := time.Now().Add(5 * time.Second)
for sc.Scan() && time.Now().Before(deadline) {
line := sc.Text()
if strings.HasPrefix(line, "data: ") && strings.Contains(line, contains) {
return line
}
}
if err := sc.Err(); err != nil {
t.Fatalf("scanner SSE: %v", err)
}
t.Fatalf("evento contendo %q não chegou", contains)
return ""
}
// hello inicial com snapshot de presença
readEvent(`"type":"hello"`)
// heartbeat de presença → broadcast presence com s2
resp := call("POST", "/api/presence", `{"session":"s2","notes":["nota1"]}`)
if resp.StatusCode != http.StatusNoContent {
t.Fatalf("presence: %s", resp.Status)
}
readEvent(`"type":"presence"`)
// presença sem session → 400
resp = call("POST", "/api/presence", `{"notes":[]}`)
if resp.StatusCode != http.StatusBadRequest {
t.Fatalf("presence sem session: %s", resp.Status)
}
resp.Body.Close()
// criar nota com session → evento saved para os ouvintes
resp = call("POST", "/api/notes", `{"title":"E","format":"txt","categories":[],"content":"x","author":"","message":"","session":"s3"}`)
if resp.StatusCode != http.StatusCreated {
t.Fatalf("create: %s", resp.Status)
}
var created noteView
decode(t, resp, &created)
ev := readEvent(`"type":"saved"`)
if !strings.Contains(ev, `"session":"s3"`) || !strings.Contains(ev, created.ID) {
t.Errorf("evento saved incompleto: %s", ev)
}
// SSE sem token → 401
resp = do(t, srv, "GET", "/api/events", "", "", "")
if resp.StatusCode != http.StatusUnauthorized {
t.Fatalf("SSE sem auth: %s", resp.Status)
}
resp.Body.Close()
}
func TestInvalidPayloads(t *testing.T) {
srv := newTestServer(t)
resp := do(t, srv, "POST", "/api/notes", `{"format":"rtf","content":"x"}`, testToken, "")
+11 -1
View File
@@ -30,6 +30,7 @@ type noteInput struct {
Author string `json:"author"`
Message string `json:"message"`
BaseHash string `json:"base_hash"` // hash em que a edição se baseia (opcional; evita sobrescrever mudanças remotas)
Session string `json:"session"` // id da sessão do cliente (presença/eventos; opcional)
}
func (in noteInput) fields() (string, string, []string, string, string, string, string) {
@@ -77,6 +78,7 @@ func (s *Server) handleCreateNote(w http.ResponseWriter, r *http.Request) {
mapStoreError(w, err)
return
}
s.publishSaved(n, author, in.Session, false)
writeJSON(w, http.StatusCreated, noteView{n, content})
}
@@ -110,14 +112,22 @@ func (s *Server) handleUpdateNote(w http.ResponseWriter, r *http.Request) {
mapStoreError(w, err)
return
}
s.publishSaved(n, author, in.Session, false)
writeJSON(w, http.StatusOK, noteView{n, content})
}
func (s *Server) handleDeleteNote(w http.ResponseWriter, r *http.Request) {
if err := s.st.Delete(r.PathValue("id")); err != nil {
id := r.PathValue("id")
n, err := s.st.Get(id)
if err != nil {
mapStoreError(w, err)
return
}
if err := s.st.Delete(id); err != nil {
mapStoreError(w, err)
return
}
s.publishSaved(n, "", "", true)
w.WriteHeader(http.StatusNoContent)
}
+251
View File
@@ -0,0 +1,251 @@
package api
import (
"encoding/json"
"fmt"
"io"
"net/http"
"sync"
"time"
"github.com/lgcosta/notas-online/internal/store"
)
// ---------------------------------------------------------------------------
// Hub SSE: transmite eventos (saved/presence) para todas as conexões /api/events
// ---------------------------------------------------------------------------
type hub struct {
mu sync.Mutex
clients map[chan string]struct{}
}
func newHub() *hub { return &hub{clients: map[chan string]struct{}{}} }
func (h *hub) register() chan string {
c := make(chan string, 32)
h.mu.Lock()
h.clients[c] = struct{}{}
h.mu.Unlock()
return c
}
func (h *hub) unregister(c chan string) {
h.mu.Lock()
delete(h.clients, c)
h.mu.Unlock()
}
// broadcast entrega sem bloquear; cliente lento perde eventos (o estado
// completo volta no próximo heartbeat/presence snapshot).
func (h *hub) broadcast(data string) {
h.mu.Lock()
defer h.mu.Unlock()
for c := range h.clients {
select {
case c <- data:
default:
}
}
}
// publish serializa v e envia ao hub.
func (s *Server) publish(v any) {
data, err := json.Marshal(v)
if err != nil {
return
}
s.hub.broadcast(string(data))
}
// ---------------------------------------------------------------------------
// Registro de presença: sessão → notas abertas (TTL por heartbeat)
// ---------------------------------------------------------------------------
const presenceTTL = 30 * time.Second
type presenceEntry struct {
notes []string
when time.Time
}
type presenceReg struct {
mu sync.Mutex
m map[string]presenceEntry
}
func newPresenceReg() *presenceReg {
return &presenceReg{m: map[string]presenceEntry{}}
}
// set grava o estado da sessão; devolve true se o conjunto de notas mudou.
func (p *presenceReg) set(session string, notes []string) bool {
p.mu.Lock()
defer p.mu.Unlock()
old, ok := p.m[session]
p.m[session] = presenceEntry{notes: notes, when: time.Now()}
if !ok {
return true
}
if len(old.notes) != len(notes) {
return true
}
for i := range notes {
if old.notes[i] != notes[i] {
return true
}
}
return false
}
func (p *presenceReg) remove(session string) bool {
p.mu.Lock()
defer p.mu.Unlock()
_, ok := p.m[session]
delete(p.m, session)
return ok
}
// snapshot devolve as sessões vivas (podando as expiradas).
func (p *presenceReg) snapshot() []presenceSession {
p.mu.Lock()
defer p.mu.Unlock()
now := time.Now()
out := make([]presenceSession, 0, len(p.m))
for sid, e := range p.m {
if now.Sub(e.when) > presenceTTL {
delete(p.m, sid)
continue
}
notes := e.notes
if notes == nil {
notes = []string{}
}
out = append(out, presenceSession{Session: sid, Notes: notes})
}
return out
}
// ---------------------------------------------------------------------------
// Eventos (payloads)
// ---------------------------------------------------------------------------
type savedEvent struct {
Type string `json:"type"` // "saved"
NoteID string `json:"note_id"`
Hash string `json:"hash"`
Title string `json:"title"`
Author string `json:"author"`
Session string `json:"session"` // quem salvou (o próprio cliente ignora)
Deleted bool `json:"deleted,omitempty"`
}
type presenceSession struct {
Session string `json:"session"`
Notes []string `json:"notes"`
}
type presenceEvent struct {
Type string `json:"type"` // "presence" | "hello"
Sessions []presenceSession `json:"sessions"`
}
// publishPresence anuncia o snapshot atual a todos os clientes SSE.
func (s *Server) publishPresence() {
s.publish(presenceEvent{Type: "presence", Sessions: s.pres.snapshot()})
}
// ---------------------------------------------------------------------------
// Handlers
// ---------------------------------------------------------------------------
// handleEvents é o stream SSE (EventSource). Auth via cookie (o EventSource
// do navegador envia cookies same-origin automaticamente).
func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request) {
flusher, ok := w.(http.Flusher)
if !ok {
writeError(w, http.StatusInternalServerError, "streaming não suportado")
return
}
session := truncate(r.URL.Query().Get("session"), 64)
ch := s.hub.register()
defer func() {
s.hub.unregister(ch)
if session != "" && s.pres.remove(session) {
s.publishPresence() // sumiu o heartbeat/sessão → atualiza todos
}
}()
h := w.Header()
h.Set("Content-Type", "text/event-stream")
h.Set("Cache-Control", "no-store")
h.Set("X-Accel-Buffering", "no")
hello, _ := json.Marshal(presenceEvent{Type: "hello", Sessions: s.pres.snapshot()})
fmt.Fprintf(w, "data: %s\n\n", hello)
flusher.Flush()
keepalive := time.NewTicker(25 * time.Second)
defer keepalive.Stop()
for {
select {
case msg := <-ch:
fmt.Fprintf(w, "data: %s\n\n", msg)
flusher.Flush()
case <-keepalive.C:
_, _ = io.WriteString(w, ": ping\n\n")
flusher.Flush()
case <-r.Context().Done():
return
}
}
}
type presenceInput struct {
Session string `json:"session"`
Notes []string `json:"notes"`
}
// handlePresence recebe o heartbeat {session, notes:[ids abertos]}.
func (s *Server) handlePresence(w http.ResponseWriter, r *http.Request) {
var in presenceInput
if err := decodeJSON(r, &in, 1<<16); err != nil {
writeError(w, http.StatusBadRequest, "corpo JSON inválido")
return
}
in.Session = truncate(in.Session, 64)
if in.Session == "" {
writeError(w, http.StatusBadRequest, "session é obrigatória")
return
}
notes := in.Notes
if len(notes) > 64 {
notes = notes[:64]
}
for _, n := range notes {
if len(n) > 32 {
writeError(w, http.StatusBadRequest, "id de nota inválido")
return
}
}
if changed := s.pres.set(in.Session, notes); changed {
s.publishPresence()
}
w.WriteHeader(http.StatusNoContent)
}
// publishSaved anuncia uma alteração de nota às demais sessões.
func (s *Server) publishSaved(n *store.Note, author, session string, deleted bool) {
s.publish(savedEvent{
Type: "saved", NoteID: n.ID, Hash: n.Hash, Title: n.Title,
Author: author, Session: session, Deleted: deleted,
})
}
func truncate(s string, n int) string {
if len(s) > n {
return s[:n]
}
return s
}
+3 -1
View File
@@ -75,7 +75,8 @@ func (s *Server) handleRevisionDiff(w http.ResponseWriter, r *http.Request) {
}
type restoreInput struct {
Author string `json:"author"`
Author string `json:"author"`
Session string `json:"session"`
}
func (s *Server) handleRestore(w http.ResponseWriter, r *http.Request) {
@@ -101,6 +102,7 @@ func (s *Server) handleRestore(w http.ResponseWriter, r *http.Request) {
mapStoreError(w, err)
return
}
s.publishSaved(n, in.Author, in.Session, false)
writeJSON(w, http.StatusOK, noteView{n, content})
}
+6
View File
@@ -29,6 +29,8 @@ type Server struct {
token string
webFS fs.FS
md goldmark.Markdown
hub *hub
pres *presenceReg
}
// New cria o servidor da API. webFS é o filesystem embutido do frontend
@@ -42,6 +44,8 @@ func New(st *store.Store, token string, webFS fs.FS) *Server {
goldmark.WithExtensions(extension.GFM),
goldmark.WithRendererOptions(htmlr.WithHardWraps()),
),
hub: newHub(),
pres: newPresenceReg(),
}
}
@@ -70,6 +74,8 @@ func (s *Server) Handler() http.Handler {
mux.HandleFunc("PUT /api/categories/{id}", s.auth(s.handleUpdateCategory))
mux.HandleFunc("DELETE /api/categories/{id}", s.auth(s.handleDeleteCategory))
mux.HandleFunc("POST /api/render", s.auth(s.handleRender))
mux.HandleFunc("GET /api/events", s.auth(s.handleEvents))
mux.HandleFunc("POST /api/presence", s.auth(s.handlePresence))
mux.HandleFunc("GET /{$}", s.serveIndex)
mux.Handle("GET /", http.FileServerFS(s.webFS))