notas-online/internal/api/presence.go

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
}