1package app23import (4 "bytes"5 "io"6 "log/slog"7 "net/http"8 "strings"9 "sync"1011 "github.com/FelineStateMachine/atlas/format/vnext"12 "github.com/FelineStateMachine/atlas/internal/logging"13)1415// Live updates (issue #5 §4.6). The registry is scanned at launch and rescanned16// on import; what a rescan changed is announced here as ready-to-swap17// partials, so a page that is open when a bundle arrives updates itself18// without a poll and without the reader doing anything.19//20// The envelope, which is contract and is stable ahead of the templates:21//22// event: catalog23// data: <hx-partial hx-target="#atlas-topbar" hx-swap="innerMorph">…</hx-partial>24//25// event: refresh26// data: <hx-partial hx-target="#atlas-shell" hx-swap="innerMorph" hx-get="/v/tunic/world"></hx-partial>27//28// Two event names and no more. `catalog` says the library's composition moved29// and carries the regions that list it. `refresh` is the one directive: the30// volume this connection is watching now serves a different build, every URL31// under its old stamp is gone, and the page has to be re-fetched whole rather32// than patched. A connection says which volume it is watching as a query33// parameter, because the server cannot otherwise know which of several open34// pages a stamp move concerns.35//36// There is no event for "a file appeared in the library directory": nothing37// watches the directory, by decision (issue #5 §2). A drop from outside38// appears at the next launch.3940// eventNames are the two events the stream carries.41const (42 eventCatalog = "catalog"43 eventRefresh = "refresh"44)4546// message is one server-sent event.47type message struct {48 name string49 body []byte5051 // volume, when set, aims the message at connections watching one volume.52 // The refresh directive is the only message that uses it.53 volume string54}5556// hub fans messages out to the open event streams. It is small on purpose:57// one process, a handful of connections, no history, no replay. A subscriber58// that cannot keep up is dropped rather than allowed to hold the publisher --59// what it missed is a catalog change, and the next one carries the whole60// state anyway.61type hub struct {62 mu sync.Mutex63 subs map[chan message]struct{}64}6566func newHub() *hub { return &hub{subs: map[chan message]struct{}{}} }6768func (h *hub) subscribe() chan message {69 stream := make(chan message, 8)70 h.mu.Lock()71 defer h.mu.Unlock()72 h.subs[stream] = struct{}{}73 return stream74}7576func (h *hub) unsubscribe(stream chan message) {77 h.mu.Lock()78 defer h.mu.Unlock()79 delete(h.subs, stream)80 close(stream)81}8283func (h *hub) publish(m message) {84 h.mu.Lock()85 defer h.mu.Unlock()86 for stream := range h.subs {87 select {88 case stream <- m:89 default:90 slog.Warn("an event stream fell behind and lost a message",91 logging.Op("events"), slog.String("event", m.name))92 }93 }94}9596// subscribers reports how many streams are open. It is here for the tests and97// for a diagnostics readout; nothing in the application branches on it.98func (h *hub) subscribers() int {99 h.mu.Lock()100 defer h.mu.Unlock()101 return len(h.subs)102}103104// announce turns what a rescan changed into the messages the stream carries.105// It is called by the import flow and by anything else that moves the library.106func (a *App) announce(changed []string) {107 if len(changed) == 0 {108 return109 }110 held := a.library()111 // The topbar carries the volume selector. An emptied library is not a112 // selector with nothing in it, it is a different page, so the shell goes113 // with the event.114 regions := []string{"topbar"}115 if len(held.order) == 0 {116 regions = append(regions, "empty-state")117 }118 body, err := renderPartials(regions, a.view(held, nil, Session{Schema: SessionSchema}))119 if err != nil {120 slog.Error("rendering a catalog event", logging.Op("events"), slog.Any("error", err))121 return122 }123 a.events.publish(message{name: eventCatalog, body: body})124125 // A volume whose serving build moved cannot be patched: every URL under126 // the old stamp is gone. The page is told to re-fetch itself.127 for _, slug := range changed {128 if refresh, ok := a.refreshMessage(held, slug); ok {129 a.events.publish(refresh)130 }131 }132 slog.Info("library changed", logging.Op("events"),133 slog.Int("volumes", len(changed)), slog.Any("changed", changed))134}135136// handleEvents holds one SSE connection open.137//138// The connection names the volume it is watching, so a stamp move is announced139// only to the pages it concerns; a connection that names nothing hears the140// catalog events and no refresh directives.141func (a *App) handleEvents(w http.ResponseWriter, r *http.Request) {142 watching := r.URL.Query().Get("volume")143 stamp := r.URL.Query().Get("stamp")144 if watching != "" && vnext.ValidSlug(watching) != nil {145 http.Error(w, "that is not a volume", http.StatusBadRequest)146 return147 }148149 w.Header().Set("Content-Type", "text/event-stream")150 w.Header().Set("Cache-Control", "no-store")151 w.Header().Set("Connection", "keep-alive")152 w.WriteHeader(http.StatusOK)153 control := http.NewResponseController(w)154155 stream := a.events.subscribe()156 defer a.events.unsubscribe(stream)157158 // A comment as soon as the subscription exists. It carries nothing, and159 // it is the only way a client -- or a test -- can know the connection is160 // live rather than merely accepted.161 if _, err := io.WriteString(w, ": atlas\n\n"); err != nil {162 return163 }164 if err := control.Flush(); err != nil {165 return166 }167 // A native file may arrive while the desktop window is still opening,168 // before this stream subscribes. The page includes the stamp it rendered;169 // comparing it after subscription closes that startup race without event170 // history or a polling loop.171 if watching != "" && stamp != "" {172 held := a.library()173 if volume, ok := held.bySlug[watching]; ok && vnext.ShortStamp(volume.Info().Stamp) != stamp {174 refresh, ok := a.refreshMessage(held, watching)175 if ok {176 if _, err := w.Write(encodeEvent(refresh)); err != nil {177 return178 }179 if err := control.Flush(); err != nil {180 return181 }182 }183 }184 }185186 for {187 select {188 case <-r.Context().Done():189 return190 case m := <-stream:191 if m.volume != "" && m.volume != watching {192 continue193 }194 if _, err := w.Write(encodeEvent(m)); err != nil {195 return196 }197 if err := control.Flush(); err != nil {198 return199 }200 }201 }202}203204func (a *App) refreshMessage(held library, slug string) (message, bool) {205 volume, serving := held.bySlug[slug]206 if !serving {207 return message{}, false208 }209 info := volume.Info()210 world := a.session(slug).World211 if _, ok := worldEntry(info, world); !ok {212 world = info.Worlds[0].Slug213 }214 where := partialTargets["shell"]215 directive := []byte("<hx-partial hx-target=\"" + where.target + "\" hx-swap=\"" + where.swap +216 "\" hx-get=\"/v/" + slug + "/" + world + "\"></hx-partial>")217 return message{name: eventRefresh, body: directive, volume: slug}, true218}219220// encodeEvent writes one message in the wire format: a name, then the body one221// data line per line of HTML, then a blank line. HTML is written as it is --222// the line splitting is the protocol's, not an escape.223func encodeEvent(m message) []byte {224 var out bytes.Buffer225 out.WriteString("event: ")226 out.WriteString(m.name)227 out.WriteString("\n")228 for _, line := range strings.Split(strings.TrimRight(string(m.body), "\n"), "\n") {229 out.WriteString("data: ")230 out.WriteString(line)231 out.WriteString("\n")232 }233 out.WriteString("\n")234 return out.Bytes()235}