Megalodon/internal/manager/hub.go

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
}