Megalodon/internal/manager/manager.go

// Package manager runs Megalodon's side of things: it keeps a connection to every server, tracks
// who is online, records events, enforces bans, runs scheduled jobs and sends notifications.
package manager

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	"git.veillax.com/Veillax/megalodon/internal/ark"
	"git.veillax.com/Veillax/megalodon/internal/config"
)

// Action is something a person or a schedule asks to be done.
type Action struct {
	Name    string   `json:"action"`
	Servers []string `json:"servers,omitempty"` // empty or ["*"]: every server
	SteamID string   `json:"steam_id,omitempty"`
	Player  string   `json:"player,omitempty"` // name, for the record
	Message string   `json:"message,omitempty"`
	Command string   `json:"command,omitempty"`
	Time    string   `json:"time,omitempty"`
	Reason  string   `json:"reason,omitempty"`
	Minutes float64  `json:"minutes,omitempty"` // restart countdown, or ban length (0 = permanent)
}

// Result is how an action went on one server.
type Result struct {
	Server string `json:"server"`
	OK     bool   `json:"ok"`
	Output string `json:"output,omitempty"`
	Error  string `json:"error,omitempty"`
}

// ActionRole is the lowest role allowed to run an action.
func ActionRole(name string) config.Role {
	switch name {
	case "broadcast", "say", "whisper", "kick", "ban", "unban", "allow", "disallow", "save":
		return config.Moderator
	}
	return config.Admin
}

type Manager struct {
	cfg     *config.Holder
	dir     string
	hub     *Hub
	players *PlayerDB
	bans    *BanDB

	mu   sync.RWMutex
	inst map[string]*Instance
	ctx  context.Context
	wg   sync.WaitGroup

	jobMu    sync.Mutex
	lastRun  map[string]time.Time
	lastFire map[string]string
	client   httpDoer
}

// New opens the data directory. Call Start to begin watching servers.
func New(h *config.Holder) (*Manager, error) {
	dir := h.Get().DataDir
	hub, err := NewHub(dir)
	if err != nil {
		return nil, err
	}
	players, err := openPlayers(dir)
	if err != nil {
		return nil, err
	}
	bans, err := openBans(dir)
	if err != nil {
		return nil, err
	}
	return &Manager{
		cfg: h, dir: dir, hub: hub, players: players, bans: bans, inst: map[string]*Instance{},
		lastRun: map[string]time.Time{}, lastFire: map[string]string{}, client: defaultClient(),
	}, nil
}

func (m *Manager) Hub() *Hub { return m.hub }

// Start begins watching every configured server and running background work until ctx ends.
func (m *Manager) Start(ctx context.Context) {
	m.ctx = ctx
	m.Reload()
	m.go_(m.scheduler)
	m.go_(m.notifier)
	m.go_(m.housekeeping)
}

func (m *Manager) go_(f func(context.Context)) {
	m.wg.Add(1)
	go func() { defer m.wg.Done(); f(m.ctx) }()
}

// Wait blocks until background work has finished (after the context given to Start is cancelled)
// and saves what is still in memory.
func (m *Manager) Wait() {
	m.wg.Wait()
	m.mu.Lock()
	for id, i := range m.inst {
		i.stop()
		delete(m.inst, id)
	}
	m.mu.Unlock()
	m.players.Flush()
	m.hub.Close()
}

// Reload makes the running servers match the configuration: new ones start, removed ones stop,
// changed ones are restarted.
func (m *Manager) Reload() {
	servers := m.cfg.Get().Servers
	want := map[string]config.Server{}
	for _, s := range servers {
		want[s.ID] = s
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	for id, i := range m.inst {
		if s, ok := want[id]; !ok || !sameServer(s, i.cfg) {
			i.stop()
			delete(m.inst, id)
		}
	}
	for _, s := range servers {
		if _, ok := m.inst[s.ID]; !ok {
			i := newInstance(m, s)
			m.inst[s.ID] = i
			i.start(m.ctx)
		}
	}
}

func sameServer(a, b config.Server) bool {
	x, _ := json.Marshal(a)
	y, _ := json.Marshal(b)
	return string(x) == string(y)
}

func (m *Manager) Instance(id string) (*Instance, bool) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	i, ok := m.inst[id]
	return i, ok
}

// Views returns every server in configuration order.
func (m *Manager) Views() []View {
	var out []View
	for _, s := range m.cfg.Get().Servers {
		if i, ok := m.Instance(s.ID); ok {
			out = append(out, i.View())
		}
	}
	return out
}

func (m *Manager) Players() []PlayerRecord { return m.players.List() }

func (m *Manager) Annotate(id, notes string, watch bool) error {
	if !ark.ValidSteamID(id) {
		return errors.New("not a Steam ID")
	}
	m.players.Annotate(id, ark.Clean(notes), watch)
	return nil
}

func (m *Manager) Bans() []Ban { return m.bans.List() }

// Audit records something done outside a server (a settings change, say).
func (m *Manager) Audit(actor, server, msg string) {
	m.hub.Publish(Event{Server: server, Type: "audit", Actor: actor, Message: msg})
}

// targets resolves server ids; empty or "*" means all.
func (m *Manager) targets(ids []string) ([]*Instance, error) {
	if len(ids) == 0 || (len(ids) == 1 && ids[0] == "*") {
		var out []*Instance
		for _, v := range m.Views() {
			if i, ok := m.Instance(v.ID); ok {
				out = append(out, i)
			}
		}
		if len(out) == 0 {
			return nil, errors.New("no servers are configured")
		}
		return out, nil
	}
	var out []*Instance
	for _, id := range ids {
		i, ok := m.Instance(id)
		if !ok {
			return nil, fmt.Errorf("unknown server %q", id)
		}
		out = append(out, i)
	}
	return out, nil
}

// Do carries out an action on its target servers, one after another, and reports each result.
// The caller has already checked that actor may do this.
func (m *Manager) Do(actor string, a Action) ([]Result, error) {
	if a.Name == "ban" || a.Name == "unban" {
		return m.banAction(actor, a)
	}
	targets, err := m.targets(a.Servers)
	if err != nil {
		return nil, err
	}
	results := make([]Result, len(targets))
	var wg sync.WaitGroup
	for k, i := range targets {
		wg.Add(1)
		go func() {
			defer wg.Done()
			out, err := i.Perform(actor, a)
			r := Result{Server: i.cfg.ID, OK: err == nil, Output: out}
			if err != nil {
				r.Error = err.Error()
			}
			results[k] = r
		}()
	}
	wg.Wait()
	return results, nil
}

func (m *Manager) banAction(actor string, a Action) ([]Result, error) {
	if a.Name == "unban" {
		return m.Unban(actor, a.SteamID)
	}
	var servers []string
	if !(len(a.Servers) == 0 || (len(a.Servers) == 1 && a.Servers[0] == "*")) {
		servers = a.Servers
	}
	return m.banOn(actor, a.SteamID, a.Player, a.Reason, time.Duration(a.Minutes*float64(time.Minute)), servers)
}

// Ban records a ban and applies it to the servers it covers (all if servers is empty). dur 0 is
// permanent. The ban is also enforced later: anyone banned who turns up on a covered server is
// removed, even if that server was offline when the ban was made.
func (m *Manager) Ban(actor, steamID, name, reason string, dur time.Duration, servers []string) error {
	_, err := m.banOn(actor, steamID, name, reason, dur, servers)
	return err
}

func (m *Manager) banOn(actor, steamID, name, reason string, dur time.Duration, servers []string) ([]Result, error) {
	if !ark.ValidSteamID(steamID) {
		return nil, errors.New("not a Steam ID")
	}
	targets, err := m.targets(servers)
	if err != nil {
		return nil, err
	}
	if name == "" {
		if r, ok := m.players.Get(steamID); ok {
			name = r.Name
		}
	}
	b := Ban{SteamID: steamID, Name: ark.Clean(name), Reason: ark.Clean(reason), By: actor, At: time.Now(), Servers: servers}
	desc := "permanently"
	if dur > 0 {
		exp := b.At.Add(dur)
		b.Expires = &exp
		desc = "for " + ark.Duration(dur)
	}
	m.bans.Add(b)
	who := steamID
	if b.Name != "" {
		who = b.Name + " (" + steamID + ")"
	}
	m.hub.Publish(Event{Type: "ban", Actor: actor, SteamID: steamID, Message: fmt.Sprintf("Banned %s %s%s", who, desc, withReason(b.Reason))})
	cmd, _ := ark.ForPlayer("BanPlayer", steamID)
	var results []Result
	for _, i := range targets {
		out, err := i.rc.Exec(cmd)
		r := Result{Server: i.cfg.ID, OK: err == nil, Output: strings.TrimSpace(out)}
		if err != nil {
			r.Error = err.Error()
		}
		results = append(results, r)
	}
	return results, nil
}

// Unban lifts a ban everywhere.
func (m *Manager) Unban(actor, steamID string) ([]Result, error) {
	if !ark.ValidSteamID(steamID) {
		return nil, errors.New("not a Steam ID")
	}
	b, _ := m.bans.Remove(steamID)
	who := steamID
	if b.Name != "" {
		who = b.Name + " (" + steamID + ")"
	}
	m.hub.Publish(Event{Type: "unban", Actor: actor, SteamID: steamID, Message: "Unbanned " + who})
	cmd, _ := ark.ForPlayer("UnbanPlayer", steamID)
	targets, _ := m.targets(nil)
	var results []Result
	for _, i := range targets {
		out, err := i.rc.Exec(cmd)
		r := Result{Server: i.cfg.ID, OK: err == nil, Output: strings.TrimSpace(out)}
		if err != nil {
			r.Error = err.Error()
		}
		results = append(results, r)
	}
	return results, nil
}

// RunMacro sends each command of a macro to the server and gathers the replies.
func (m *Manager) RunMacro(actor, server string, mc config.Macro) ([]Result, error) {
	i, ok := m.Instance(server)
	if !ok {
		return nil, fmt.Errorf("unknown server %q", server)
	}
	m.Audit(actor, server, "ran macro "+mc.Name)
	var out []Result
	for _, c := range mc.Commands {
		reply, err := i.Exec(c)
		r := Result{Server: server, OK: err == nil, Output: strings.TrimSpace(reply)}
		if err != nil {
			r.Error = err.Error()
		}
		out = append(out, r)
		if err != nil {
			break
		}
	}
	return out, nil
}

// ---- background work ----

// housekeeping flushes the player list and lifts expired bans.
func (m *Manager) housekeeping(ctx context.Context) {
	t := time.NewTicker(30 * time.Second)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			m.players.Flush()
			for _, b := range m.bans.Expired(time.Now()) {
				m.Unban("megalodon", b.SteamID)
			}
		}
	}
}

// scheduler runs the configured jobs.
func (m *Manager) scheduler(ctx context.Context) {
	t := time.NewTicker(15 * time.Second)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case now := <-t.C:
			for _, j := range m.cfg.Get().Jobs {
				if j.Enabled && m.due(j, now) {
					go m.RunJob(j, "job: "+j.Name)
				}
			}
		}
	}
}

var dayNames = [...]string{"sun", "mon", "tue", "wed", "thu", "fri", "sat"}

func (m *Manager) due(j config.Job, now time.Time) bool {
	m.jobMu.Lock()
	defer m.jobMu.Unlock()
	if j.Every != "" {
		d, err := time.ParseDuration(j.Every)
		if err != nil {
			return false
		}
		last, ok := m.lastRun[j.ID]
		if !ok { // first sight: count from now rather than firing at once
			m.lastRun[j.ID] = now
			return false
		}
		if now.Sub(last) >= d {
			m.lastRun[j.ID] = now
			return true
		}
		return false
	}
	if len(j.Days) > 0 {
		ok := false
		for _, d := range j.Days {
			ok = ok || d == dayNames[now.Weekday()]
		}
		if !ok {
			return false
		}
	}
	minute := now.Format("2006-01-02 15:04")
	for _, at := range j.At {
		if at == now.Format("15:04") && m.lastFire[j.ID] != minute {
			m.lastFire[j.ID] = minute
			return true
		}
	}
	return false
}

// RunJob runs a job now.
func (m *Manager) RunJob(j config.Job, actor string) []Result {
	a := Action{Name: j.Action, Servers: j.Servers, Message: j.Message, Reason: j.Reason}
	switch j.Action {
	case "command":
		a.Name, a.Command = "exec", j.Command
	case "restart":
		d := 5 * time.Minute
		if j.Countdown != "" {
			d, _ = time.ParseDuration(j.Countdown)
		}
		a.Minutes = d.Minutes()
	}
	results, err := m.Do(actor, a)
	if err != nil {
		m.hub.Publish(Event{Type: "error", Actor: actor, Message: "Job " + j.Name + " failed: " + err.Error()})
		return nil
	}
	failed := 0
	for _, r := range results {
		if !r.OK {
			failed++
		}
	}
	msg := fmt.Sprintf("Ran job %q on %d server(s)", j.Name, len(results))
	typ := "schedule"
	if failed > 0 {
		msg += fmt.Sprintf(", %d failed", failed)
		typ = "error"
	}
	m.hub.Publish(Event{Type: typ, Actor: actor, Message: msg})
	return results
}

// Metrics renders the current state in Prometheus text format.
func (m *Manager) Metrics() string {
	var sb strings.Builder
	views := m.Views()
	sort.Slice(views, func(a, b int) bool { return views[a].ID < views[b].ID })
	sb.WriteString("# TYPE megalodon_server_up gauge\n# TYPE megalodon_players gauge\n# TYPE megalodon_players_max gauge\n# TYPE megalodon_rcon_rtt_ms gauge\n# TYPE megalodon_uptime_seconds gauge\n# TYPE megalodon_restart_pending gauge\n")
	for _, v := range views {
		l := fmt.Sprintf(`{server=%q}`, v.ID)
		up := 0
		if v.Online {
			up = 1
		}
		fmt.Fprintf(&sb, "megalodon_server_up%s %d\nmegalodon_players%s %d\nmegalodon_rcon_rtt_ms%s %d\n", l, up, l, len(v.Players), l, v.RTTMS)
		if v.Info != nil {
			fmt.Fprintf(&sb, "megalodon_players_max%s %d\n", l, v.Info.MaxPlayers)
		}
		if v.Since != nil {
			fmt.Fprintf(&sb, "megalodon_uptime_seconds%s %d\n", l, int(time.Since(*v.Since).Seconds()))
		}
		pending := 0
		if v.Restart != nil {
			pending = 1
		}
		fmt.Fprintf(&sb, "megalodon_restart_pending%s %d\n", l, pending)
	}
	return sb.String()
}