package server import ( "encoding/json" "sync" ) // Event is one push notification sent to clients over SSE. type Event struct { Type string `json:"type"` DatabaseID string `json:"database_id,omitempty"` Name string `json:"name,omitempty"` Version int64 `json:"version,omitempty"` SHA256 string `json:"sha256,omitempty"` By string `json:"by,omitempty"` } const ( EventUpdated = "db-updated" EventChanged = "db-list-changed" ) // subBuffer is how many events a slow subscriber may fall behind before its // stream is dropped. A dropped client reconnects and resyncs from scratch, // so losing events is safe; blocking the publisher would not be. const subBuffer = 16 type subscriber struct { userID string ch chan Event } // Hub fans events out to the connected clients of specific users. type Hub struct { mu sync.Mutex next int subs map[int]*subscriber } func NewHub() *Hub { return &Hub{subs: map[int]*subscriber{}} } // Subscribe registers a stream for one user and returns its id and channel. func (h *Hub) Subscribe(userID string) (int, <-chan Event) { h.mu.Lock() defer h.mu.Unlock() h.next++ s := &subscriber{userID: userID, ch: make(chan Event, subBuffer)} h.subs[h.next] = s return h.next, s.ch } // Unsubscribe removes a stream and closes its channel. func (h *Hub) Unsubscribe(id int) { h.mu.Lock() defer h.mu.Unlock() if s, ok := h.subs[id]; ok { delete(h.subs, id) close(s.ch) } } // Publish delivers ev to every stream of every listed user. A stream that is // not keeping up is dropped rather than waited for. func (h *Hub) Publish(userIDs []string, ev Event) { want := make(map[string]bool, len(userIDs)) for _, id := range userIDs { want[id] = true } h.mu.Lock() defer h.mu.Unlock() for id, s := range h.subs { if !want[s.userID] { continue } select { case s.ch <- ev: default: delete(h.subs, id) close(s.ch) } } } // Count reports the number of live streams, for the health endpoint. func (h *Hub) Count() int { h.mu.Lock() defer h.mu.Unlock() return len(h.subs) } func (e Event) encode() []byte { b, _ := json.Marshal(e) return b }