120 lines
3.2 KiB
Go
120 lines
3.2 KiB
Go
package jmap
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// stateChange is RFC 8620 §7.2's StateChange object, sent as one SSE "data:" line per
|
|
// push event.
|
|
type stateChange struct {
|
|
Type string `json:"@type"`
|
|
Changed map[string]map[string]any `json:"changed"`
|
|
}
|
|
|
|
// currentStates returns the requested types' current state strings for mbox — only
|
|
// Email and Mailbox are tracked by this server (see state.go); an unrecognized type
|
|
// name in the request is silently skipped rather than erroring, matching JMAP's own
|
|
// "ignore unknown capabilities" posture elsewhere.
|
|
func currentStates(b *Backend, mboxID int64, types []string) (map[string]any, error) {
|
|
out := map[string]any{}
|
|
for _, t := range types {
|
|
switch t {
|
|
case "Email":
|
|
state, err := b.DB.MessagesState(mboxID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out["Email"] = state
|
|
case "Mailbox":
|
|
state, err := b.DB.FoldersState(mboxID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out["Mailbox"] = state
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// handleEventSource is GET /jmap/eventsource (RFC 8620 §7.3) — this server's push
|
|
// transport. Pushes account-wide, not per-folder like IMAP IDLE/webmail's own SSE
|
|
// (internal/webui's mailStream): subscribes to notify.Bus's account-wide sentinel key
|
|
// (see notify.Bus.PublishAccountWide) instead of fanning out across every folder,
|
|
// since a JMAP client's changed-state query is itself account-wide, not per-folder.
|
|
func (b *Backend) handleEventSource(w http.ResponseWriter, r *http.Request) {
|
|
mbox := mailboxFromContext(r)
|
|
|
|
types := []string{"Email", "Mailbox"}
|
|
if raw := r.URL.Query().Get("types"); raw != "" && raw != "*" {
|
|
types = strings.Split(raw, ",")
|
|
}
|
|
closeAfterState := r.URL.Query().Get("closeafter") == "state"
|
|
ping := 30 * time.Second
|
|
if raw := r.URL.Query().Get("ping"); raw != "" {
|
|
if secs, err := strconv.Atoi(raw); err == nil && secs > 0 {
|
|
ping = time.Duration(secs) * time.Second
|
|
}
|
|
}
|
|
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
w.WriteHeader(http.StatusOK)
|
|
flusher.Flush()
|
|
|
|
sendState := func() bool {
|
|
changed, err := currentStates(b, mbox.ID, types)
|
|
if err != nil {
|
|
return true // keep the connection open — a transient DB error shouldn't kill it
|
|
}
|
|
body, err := json.Marshal(stateChange{
|
|
Type: "StateChange",
|
|
Changed: map[string]map[string]any{strconv.FormatInt(mbox.ID, 10): changed},
|
|
})
|
|
if err != nil {
|
|
return true
|
|
}
|
|
if _, err := fmt.Fprintf(w, "event: state\ndata: %s\n\n", body); err != nil {
|
|
return false
|
|
}
|
|
flusher.Flush()
|
|
return true
|
|
}
|
|
|
|
ch, unsubscribe := b.Notify.Subscribe(mbox.ID, "")
|
|
defer unsubscribe()
|
|
|
|
keepalive := time.NewTicker(ping)
|
|
defer keepalive.Stop()
|
|
|
|
ctx := r.Context()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ch:
|
|
if !sendState() {
|
|
return
|
|
}
|
|
if closeAfterState {
|
|
return
|
|
}
|
|
case <-keepalive.C:
|
|
if _, err := fmt.Fprint(w, ": keepalive\n\n"); err != nil {
|
|
return
|
|
}
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|