package manager import ( "bufio" "encoding/json" "os" "path/filepath" "strings" "sync" "time" ) // Event is one thing that happened: a player joined, a server went down, someone ran a command. type Event struct { ID int64 `json:"id"` Time time.Time `json:"time"` Server string `json:"server,omitempty"` Type string `json:"type"` // online, offline, join, leave, chat, audit, restart, start, ban, unban, kick, filter, watch, schedule, error Actor string `json:"actor,omitempty"` Message string `json:"message"` Detail string `json:"detail,omitempty"` SteamID string `json:"steam_id,omitempty"` } const ( ringMax = 3000 rotateAfter = 20 << 20 ) // Hub records events (in memory and in a JSON-lines file) and hands them to anyone listening. type Hub struct { mu sync.Mutex seq int64 ring []Event subs map[chan Event]struct{} path string f *os.File written int64 } // NewHub opens dir/events.jsonl, loading the most recent events from it. func NewHub(dir string) (*Hub, error) { if err := os.MkdirAll(dir, 0o755); err != nil { return nil, err } h := &Hub{path: filepath.Join(dir, "events.jsonl"), subs: map[chan Event]struct{}{}} if f, err := os.Open(h.path); err == nil { sc := bufio.NewScanner(f) sc.Buffer(make([]byte, 64<<10), 1<<20) for sc.Scan() { var e Event if json.Unmarshal(sc.Bytes(), &e) == nil && e.ID > 0 { h.ring = append(h.ring, e) if len(h.ring) > ringMax { h.ring = h.ring[len(h.ring)-ringMax:] } if e.ID > h.seq { h.seq = e.ID } } } f.Close() } if err := h.open(); err != nil { return nil, err } return h, nil } func (h *Hub) open() error { f, err := os.OpenFile(h.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600) if err != nil { return err } st, _ := f.Stat() if st != nil { h.written = st.Size() } h.f = f return nil } func (h *Hub) Close() { h.mu.Lock() defer h.mu.Unlock() if h.f != nil { h.f.Close() h.f = nil } } // Publish stamps and stores an event and returns it as stored. func (h *Hub) Publish(e Event) Event { h.mu.Lock() defer h.mu.Unlock() h.seq++ e.ID = h.seq if e.Time.IsZero() { e.Time = time.Now() } h.ring = append(h.ring, e) if len(h.ring) > ringMax { h.ring = append([]Event(nil), h.ring[len(h.ring)-ringMax:]...) } if h.f != nil { if b, err := json.Marshal(e); err == nil { n, _ := h.f.Write(append(b, '\n')) h.written += int64(n) if h.written > rotateAfter { h.f.Close() os.Rename(h.path, h.path+".1") h.open() } } } for ch := range h.subs { select { case ch <- e: default: // a slow listener misses events rather than holding everyone up } } return e } // Subscribe returns a channel of new events and a function that stops it. func (h *Hub) Subscribe() (<-chan Event, func()) { ch := make(chan Event, 64) h.mu.Lock() h.subs[ch] = struct{}{} h.mu.Unlock() return ch, func() { h.mu.Lock() delete(h.subs, ch) h.mu.Unlock() } } // Filter selects events for Recent. type Filter struct { Server string Types []string Query string // case-insensitive text in the message or actor Before int64 // only events with a lower id (for paging back) Limit int } // Recent returns matching events, newest first. func (h *Hub) Recent(f Filter) []Event { if f.Limit <= 0 || f.Limit > 1000 { f.Limit = 200 } q := strings.ToLower(f.Query) want := map[string]bool{} for _, t := range f.Types { want[t] = true } h.mu.Lock() defer h.mu.Unlock() var out []Event for i := len(h.ring) - 1; i >= 0 && len(out) < f.Limit; i-- { e := h.ring[i] if f.Before > 0 && e.ID >= f.Before { continue } if f.Server != "" && e.Server != f.Server { continue } if len(want) > 0 && !want[e.Type] { continue } if q != "" && !strings.Contains(strings.ToLower(e.Message+" "+e.Actor), q) { continue } out = append(out, e) } return out }