package server import ( "sort" "sync" "time" "g.fc.run/misc/byre/internal/api" ) type clientKey struct{ login, host, commit string } type hub struct { mu sync.Mutex subs map[chan api.Event]api.ConnectedClient lastAction map[clientKey]time.Time lastNotified map[clientKey]time.Time } func newHub() *hub { return &hub{subs: map[chan api.Event]api.ConnectedClient{}, lastAction: map[clientKey]time.Time{}, lastNotified: map[clientKey]time.Time{}} } func (h *hub) acted(login, host, commit string) bool { h.mu.Lock() defer h.mu.Unlock() k := clientKey{login, host, commit} now := time.Now() h.lastAction[k] = now prev, ok := h.lastNotified[k] if ok && now.Sub(prev) < time.Second { return false } h.lastNotified[k] = now return true } func (h *hub) subscribe(client api.ConnectedClient) chan api.Event { h.mu.Lock() defer h.mu.Unlock() ch := make(chan api.Event, 256) h.subs[ch] = client return ch } func (h *hub) unsubscribe(ch chan api.Event) { h.mu.Lock() defer h.mu.Unlock() _, ok := h.subs[ch] if !ok { return } delete(h.subs, ch) close(ch) } func (h *hub) publish(ev api.Event) { h.mu.Lock() defer h.mu.Unlock() for ch := range h.subs { select { case ch <- ev: default: delete(h.subs, ch) close(ch) } } } func (h *hub) count() int { h.mu.Lock() defer h.mu.Unlock() return len(h.subs) } func (h *hub) clients() []api.ConnectedClient { h.mu.Lock() defer h.mu.Unlock() oldest := map[clientKey]api.ConnectedClient{} for _, c := range h.subs { k := clientKey{c.Login, c.Host, c.Commit} prev, ok := oldest[k] if ok && !c.ConnectedAt.Before(prev.ConnectedAt) { continue } oldest[k] = c } out := []api.ConnectedClient{} for k, c := range oldest { at, ok := h.lastAction[k] if ok { c.LastActionAt = &at } out = append(out, c) } sort.Slice(out, func(i, j int) bool { return out[i].ConnectedAt.Before(out[j].ConnectedAt) }) return out }