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 } // SSE é um stream longo; o WriteTimeout global fecharia a conexão. Renovamos // o deadline a cada keepalive via ResponseController (Go 1.20+). rc := http.NewResponseController(w) _ = rc.SetWriteDeadline(time.Time{}) 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: _ = rc.SetWriteDeadline(time.Time{}) _, _ = 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(w, r, &in, 1<<16); err != nil { 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 }