Realtime push (IMAP IDLE → SSE) + delta-sync latency fixes

server/cmd/server/main.go

@@ -16,6 +16,7 @@ import (
 	"quicksilver/server/internal/config"
 	apihttp "quicksilver/server/internal/http"
 	applog "quicksilver/server/internal/log"
+	"quicksilver/server/internal/realtime"
 	"quicksilver/server/internal/session"
 	"quicksilver/server/internal/smtp"
 )
@@ -60,6 +61,7 @@ func main() {
 	}
 	sessions := session.NewStore(bgCtx, sealer, cfg.SessionIdleTTL, cfg.SessionSweepInt, cfg.IMAPTimeout, logger)
 	sender := smtp.New(cfg.SMTPTimeout)
+	hub := realtime.NewHub(logger)
 
 	router := apihttp.NewRouter(apihttp.Deps{
 		Config:       cfg,
@@ -70,6 +72,7 @@ func main() {
 		Issuer:       issuer,
 		Sealer:       sealer,
 		Sender:       sender,
+		Hub:          hub,
 		RateLimitCtx: bgCtx,
 	})
 

server/internal/http/handlers/events.go

@@ -0,0 +1,131 @@
+package handlers
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"strings"
+	"time"
+
+	"quicksilver/server/internal/http/middleware"
+	"quicksilver/server/internal/httpx"
+	"quicksilver/server/internal/imap"
+	"quicksilver/server/internal/realtime"
+	"quicksilver/server/internal/session"
+)
+
+// Events serves the realtime change-notification stream over Server-Sent Events
+// (proposal §6, Phase 4). The browser opens one EventSource; the gateway holds
+// an IMAP IDLE connection per watched mailbox and pushes a tiny "changed" event
+// whenever that mailbox changes, prompting the client to run a delta sync.
+type Events struct {
+	Sessions  *session.Store
+	Hub       *realtime.Hub
+	Logger    *slog.Logger
+	Heartbeat time.Duration // comment-ping interval to keep the stream alive
+}
+
+// Stream is GET /api/v1/events. Auth is via the standard bearer token, which
+// RequireSession also accepts as an access_token query param because the
+// browser EventSource API cannot set request headers.
+//
+// Query params:
+//   - mailbox: comma-separated mailbox names to watch (default INBOX)
+func (h *Events) Stream(w http.ResponseWriter, r *http.Request) {
+	sess, ok := middleware.SessionFrom(r.Context())
+	if !ok {
+		httpx.WriteError(w, r, h.Logger, httpx.NewAPIError(http.StatusUnauthorized, httpx.CodeUnauthorized, "no session", nil))
+		return
+	}
+
+	mailboxes := parseMailboxes(r.URL.Query().Get("mailbox"))
+	if len(mailboxes) == 0 {
+		mailboxes = []string{"INBOX"}
+	}
+
+	rc := http.NewResponseController(w)
+	// SSE streams stay open far longer than the server's WriteTimeout; clear the
+	// per-connection write deadline so it isn't killed mid-stream.
+	if err := rc.SetWriteDeadline(time.Time{}); err != nil {
+		h.Logger.Warn("events: clear write deadline", "err", err)
+	}
+
+	h2 := w.Header()
+	h2.Set("Content-Type", "text/event-stream")
+	h2.Set("Cache-Control", "no-cache")
+	h2.Set("Connection", "keep-alive")
+	h2.Set("X-Accel-Buffering", "no") // disable proxy buffering (nginx)
+	w.WriteHeader(http.StatusOK)
+
+	dial := func(ctx context.Context) (*imap.Client, error) {
+		return h.Sessions.DialIMAP(ctx, sess)
+	}
+	sub, cleanup := h.Hub.Subscribe(sess.ID, dial, mailboxes)
+	defer cleanup()
+
+	// Open the stream with a comment so the browser fires `onopen` promptly.
+	if _, err := fmt.Fprint(w, ": connected\n\n"); err != nil {
+		return
+	}
+	_ = rc.Flush()
+
+	heartbeat := h.Heartbeat
+	if heartbeat <= 0 {
+		heartbeat = 25 * time.Second
+	}
+	hb := time.NewTicker(heartbeat)
+	defer hb.Stop()
+
+	ctx := r.Context()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case ev := <-sub.C:
+			b, err := json.Marshal(ev)
+			if err != nil {
+				continue
+			}
+			if _, err := fmt.Fprintf(w, "event: changed\ndata: %s\n\n", b); err != nil {
+				return
+			}
+			if err := rc.Flush(); err != nil {
+				return
+			}
+		case <-hb.C:
+			// Keep the session warm while the user is actively connected, and
+			// keep idle proxies from closing the stream.
+			sess.Touch()
+			if _, err := fmt.Fprint(w, ": ping\n\n"); err != nil {
+				return
+			}
+			if err := rc.Flush(); err != nil {
+				return
+			}
+		}
+	}
+}
+
+// parseMailboxes splits a comma-separated mailbox list, trimming blanks and
+// capping the count so one connection can't spin up an unbounded number of IDLE
+// connections.
+func parseMailboxes(s string) []string {
+	if s == "" {
+		return nil
+	}
+	parts := strings.Split(s, ",")
+	out := make([]string, 0, len(parts))
+	for _, p := range parts {
+		p = strings.TrimSpace(p)
+		if p == "" {
+			continue
+		}
+		out = append(out, p)
+		if len(out) >= 8 {
+			break
+		}
+	}
+	return out
+}

server/internal/http/handlers/mailboxes.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
+	"strings"
 
 	"github.com/go-chi/chi/v5"
 
@@ -63,14 +64,72 @@ func (h *Mailboxes) ListMessages(w http.ResponseWriter, r *http.Request) {
 		httpx.WriteError(w, r, h.Logger, httpx.NewAPIError(http.StatusBadGateway, httpx.CodeUpstreamFailed, "imap connect", err))
 		return
 	}
-	envelopes, total, err := c.ListMessages(r.Context(), mailbox, limit, uint32(before))
+	envelopes, total, uidvalidity, err := c.ListMessages(r.Context(), mailbox, limit, uint32(before))
 	if err != nil {
 		httpx.WriteError(w, r, h.Logger, httpx.NewAPIError(http.StatusBadGateway, httpx.CodeUpstreamFailed, "list messages", err))
 		return
 	}
-	resp := map[string]any{"messages": envelopes, "total": total}
+	resp := map[string]any{"messages": envelopes, "total": total, "uidvalidity": uidvalidity}
 	if len(envelopes) > 0 {
 		resp["next_before"] = envelopes[len(envelopes)-1].UID
 	}
 	httpx.WriteJSON(w, http.StatusOK, resp)
 }
+
+// Changes returns an incremental-sync delta for the named mailbox (proposal §6),
+// so the client refreshes by fetching only what changed rather than re-listing.
+//
+// Query params:
+//   - uidvalidity (the client's cached UIDVALIDITY; 0/absent on first sync)
+//   - known       (comma-separated UIDs the client currently holds)
+//   - limit       (1..200, default 50 — caps how many new envelopes are returned)
+func (h *Mailboxes) Changes(w http.ResponseWriter, r *http.Request) {
+	sess, ok := middleware.SessionFrom(r.Context())
+	if !ok {
+		httpx.WriteError(w, r, h.Logger, httpx.NewAPIError(http.StatusUnauthorized, httpx.CodeUnauthorized, "no session", nil))
+		return
+	}
+	mailbox, err := url.PathUnescape(chi.URLParam(r, "mailbox"))
+	if err != nil || mailbox == "" {
+		httpx.WriteError(w, r, h.Logger, httpx.NewAPIError(http.StatusBadRequest, httpx.CodeBadRequest, "invalid mailbox name", err))
+		return
+	}
+	q := r.URL.Query()
+	uidvalidity, _ := strconv.ParseUint(q.Get("uidvalidity"), 10, 32)
+	limit, _ := strconv.Atoi(q.Get("limit"))
+	known := parseUIDList(q.Get("known"))
+
+	c, err := h.Sessions.IMAPFor(r.Context(), sess)
+	if err != nil {
+		httpx.WriteError(w, r, h.Logger, httpx.NewAPIError(http.StatusBadGateway, httpx.CodeUpstreamFailed, "imap connect", err))
+		return
+	}
+	delta, err := c.MailboxChanges(r.Context(), mailbox, uint32(uidvalidity), known, limit)
+	if err != nil {
+		httpx.WriteError(w, r, h.Logger, httpx.NewAPIError(http.StatusBadGateway, httpx.CodeUpstreamFailed, "mailbox changes", err))
+		return
+	}
+	httpx.WriteJSON(w, http.StatusOK, delta)
+}
+
+// parseUIDList parses a comma-separated list of UIDs, skipping any malformed or
+// zero entries. Caps the input to a sane bound so a runaway query string can't
+// force an unbounded FETCH.
+func parseUIDList(s string) []uint32 {
+	if s == "" {
+		return nil
+	}
+	parts := strings.Split(s, ",")
+	if len(parts) > 1000 {
+		parts = parts[:1000]
+	}
+	out := make([]uint32, 0, len(parts))
+	for _, p := range parts {
+		n, err := strconv.ParseUint(strings.TrimSpace(p), 10, 32)
+		if err != nil || n == 0 {
+			continue
+		}
+		out = append(out, uint32(n))
+	}
+	return out
+}

server/internal/http/handlers/mailboxes_test.go

@@ -0,0 +1,48 @@
+package handlers
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestParseUIDList(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want []uint32
+	}{
+		{"empty", "", nil},
+		{"single", "42", []uint32{42}},
+		{"multiple", "1,2,3", []uint32{1, 2, 3}},
+		{"whitespace", " 1 , 2 ,3 ", []uint32{1, 2, 3}},
+		{"skips zero", "0,5,0,7", []uint32{5, 7}},
+		{"skips malformed", "1,abc,3,,5", []uint32{1, 3, 5}},
+		{"all invalid", "x,y,z", []uint32{}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := parseUIDList(tt.in)
+			if len(got) == 0 && len(tt.want) == 0 {
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("parseUIDList(%q) = %v, want %v", tt.in, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestParseUIDListCapsInput(t *testing.T) {
+	// Build a 1500-entry list; parser must cap the parsed set at 1000 so a
+	// runaway query string can't force an unbounded FETCH.
+	in := ""
+	for i := 0; i < 1500; i++ {
+		if i > 0 {
+			in += ","
+		}
+		in += "1"
+	}
+	if got := len(parseUIDList(in)); got > 1000 {
+		t.Errorf("parseUIDList did not cap input: got %d entries, want <= 1000", got)
+	}
+}

server/internal/http/middleware/auth.go

@@ -49,8 +49,12 @@ func RequireSession(issuer *auth.Issuer, store *session.Store) func(http.Handler
 func bearerToken(r *http.Request) string {
 	h := r.Header.Get("Authorization")
 	const prefix = "Bearer "
-	if !strings.HasPrefix(h, prefix) {
-		return ""
+	if strings.HasPrefix(h, prefix) {
+		return strings.TrimSpace(h[len(prefix):])
 	}
-	return strings.TrimSpace(h[len(prefix):])
+	// Fall back to the access_token query param. The browser EventSource API
+	// (used by the SSE /events stream) cannot set an Authorization header, so it
+	// passes the token in the query string. Safe here because the access log
+	// records only r.URL.Path, never the raw query.
+	return strings.TrimSpace(r.URL.Query().Get("access_token"))
 }

server/internal/http/router.go

@@ -11,6 +11,7 @@ import (
 	"quicksilver/server/internal/config"
 	"quicksilver/server/internal/http/handlers"
 	"quicksilver/server/internal/http/middleware"
+	"quicksilver/server/internal/realtime"
 	"quicksilver/server/internal/session"
 	"quicksilver/server/internal/smtp"
 )
@@ -25,6 +26,7 @@ type Deps struct {
 	Issuer   *auth.Issuer
 	Sealer   *session.Sealer
 	Sender   *smtp.Sender
+	Hub      *realtime.Hub
 	// RateLimitCtx scopes the rate limiter's background reaper. When the
 	// context is cancelled, the limiter stops reaping idle IP entries.
 	RateLimitCtx context.Context
@@ -47,6 +49,7 @@ func NewRouter(d Deps) http.Handler {
 	authH := &handlers.Auth{Sessions: d.Sessions, Issuer: d.Issuer, Logger: d.Logger}
 	mboxH := &handlers.Mailboxes{Sessions: d.Sessions, Logger: d.Logger}
 	msgH := &handlers.Messages{Sessions: d.Sessions, Sealer: d.Sealer, Sender: d.Sender, Logger: d.Logger}
+	eventsH := &handlers.Events{Sessions: d.Sessions, Hub: d.Hub, Logger: d.Logger}
 
 	requireSession := middleware.RequireSession(d.Issuer, d.Sessions)
 
@@ -63,10 +66,15 @@ func NewRouter(d Deps) http.Handler {
 			r.Use(requireSession)
 			r.Get("/mailboxes", mboxH.List)
 			r.Get("/mailboxes/{mailbox}/messages", mboxH.ListMessages)
+			r.Get("/mailboxes/{mailbox}/changes", mboxH.Changes)
 			r.Get("/mailboxes/{mailbox}/messages/{uid}", msgH.Get)
 			r.Patch("/mailboxes/{mailbox}/messages/{uid}/flags", msgH.SetFlags)
 			r.Delete("/mailboxes/{mailbox}/messages/{uid}", msgH.Delete)
 			r.Post("/messages", msgH.Send)
+
+			// Realtime change stream (SSE). RequireSession also accepts the JWT
+			// as an access_token query param since EventSource can't set headers.
+			r.Get("/events", eventsH.Stream)
 		})
 	})