Maven/cmd/mavpoll/main.go

// mavpoll — the env poller module.
//
// maven doesn't collect metrics; netdata and uptime-kuma already do, tuned to
// the box. This is a thin adapter: it reads their alarms/status and writes
// `facts (kind=env, source=poll:*)` through core's IPC socket. Key-free,
// restart-free, fail-independent — a crashing poller can't touch the store key
// (it never had it), worst case a stale env fact until the next tick.
//
// Four sources, each its own provenance (the loop's rules trust source):
//   - netdata  → poll:netdata      resource alarms (disk/mem/cert/temp)
//   - kuma     → poll:uptimekuma   service up/down (the source of truth for it)
//   - zenmoney → poll:zenmoney     spending/income totals (Vikunja #125)
//   - wireguard → infer:wg         latest handshake, the presence signal
//
// The zenmoney source is why the token lives HERE and not in core: the poller
// already owns every other third-party credential, it holds no store key, and
// core never needs to know an account exists to answer a question about a fact
// the poller wrote. It is off unless -zenmoney-token-file is given.
//
// Netdata needs no auth over the wg-fronted net. Kuma's /metrics needs an API
// key (basic-auth); without -kuma the whole kuma path is skipped (netdata-only
// still lights up an end-to-end nudge).
//
// Append-only discipline: a fact is written only when its value CHANGED vs the
// latest for that key+source. A poller that wrote every 60s would churn the
// facts table for nothing; the store is the audit trail, not a metrics sink.
package main

import (
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"os/exec"
	"os/signal"
	"regexp"
	"strconv"
	"strings"
	"syscall"
	"time"

	"github.com/kami/maven/internal/ipc"
	"github.com/kami/maven/internal/zenmoney"
)

func main() {
	if err := run(os.Args[1:]); err != nil {
		fmt.Fprintln(os.Stderr, "mavpoll:", err)
		os.Exit(1)
	}
}

func run(args []string) error {
	fs := flag.NewFlagSet("mavpoll", flag.ContinueOnError)
	socket := fs.String("socket", "", "core IPC socket path (required)")
	netdataURL := fs.String("netdata", "http://127.0.0.1:19999", "netdata base URL ('' to disable)")
	kumaURL := fs.String("kuma", "", "uptime-kuma metrics URL, e.g. http://127.0.0.1:3001/metrics ('' to disable)")
	kumaKey := fs.String("kuma-key", "", "uptime-kuma API key (basic-auth username)")
	zenTokenFile := fs.String("zenmoney-token-file", "", "file holding the zenmoney API token ('' disables money tracking)")
	zenURL := fs.String("zenmoney-url", zenmoney.DefaultBaseURL, "zenmoney API base URL (tests/self-hosted proxies)")
	zenInterval := fs.Duration("zenmoney-interval", time.Hour, "how often to read zenmoney (money does not move every minute)")
	wgIface := fs.String("wg", "", "wireguard interface for the presence signal, e.g. wg0 or 'all' ('' to disable)")
	wgCmd := fs.String("wg-cmd", "wg", "wg binary (use e.g. 'sudo wg' if the poller lacks CAP_NET_ADMIN)")
	interval := fs.Duration("interval", 60*time.Second, "poll cadence")
	timeout := fs.Duration("timeout", 8*time.Second, "per-request HTTP timeout")
	if err := fs.Parse(args); err != nil {
		return err
	}
	if *socket == "" {
		return fmt.Errorf("-socket is required")
	}
	if *netdataURL == "" && *kumaURL == "" && *wgIface == "" && *zenTokenFile == "" {
		return fmt.Errorf("nothing to poll: set -netdata, -kuma, -wg and/or -zenmoney-token-file")
	}
	// A bad duration or an empty -wg-cmd used to get past start and kill the
	// poller on the first tick — time.NewTicker panics on a non-positive
	// interval, and pollWg indexed field 0 of an empty command. A zero -timeout
	// is worse than a crash: http.Client reads it as "no deadline", so one
	// wedged source stalls every other source behind it forever. Refuse all
	// three here, where the operator sees the message.
	if *interval <= 0 {
		return fmt.Errorf("-interval must be positive, got %s", *interval)
	}
	if *timeout <= 0 {
		return fmt.Errorf("-timeout must be positive, got %s", *timeout)
	}
	if *wgIface != "" && strings.TrimSpace(*wgCmd) == "" {
		return fmt.Errorf("-wg-cmd is empty but -wg is set")
	}

	zen, err := newZenClient(*zenTokenFile, *zenURL, *timeout)
	if err != nil {
		return err
	}

	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	core, err := ipc.DialWait(*socket, coreDialWait)
	if err != nil {
		return err
	}
	defer core.Close()

	p := &poller{
		core:       core,
		http:       &http.Client{Timeout: *timeout},
		netdataURL: strings.TrimRight(*netdataURL, "/"),
		kumaURL:    *kumaURL,
		kumaKey:    *kumaKey,
		wgIface:    *wgIface,
		wgCmd:      *wgCmd,
		zen:        zen,
		zenEvery:   *zenInterval,
	}

	// The token is never logged, not even its length.
	log.Printf("mavpoll: polling every %s (netdata=%q kuma=%q wg=%q zenmoney=%v every %s)",
		*interval, *netdataURL, *kumaURL, *wgIface, zen != nil, *zenInterval)
	p.loop(ctx, *interval)
	return nil
}

// coreDialWait — how long to wait for core's socket at start. The poller and
// core come up together under compose, so a cold start is a wait, not a failure.
const coreDialWait = 60 * time.Second

// zenTimeoutFactor — the zenmoney client gets a longer deadline than the other
// sources. A diff call walks his whole transaction history, where netdata and
// kuma answer from memory.
const zenTimeoutFactor = 3

// newZenClient builds the money client, or nil when no token file was given.
//
// The token is read from a file, never taken as a flag value: an argv token is
// visible in `ps` to every user on the box and lands in the compose file and
// the shell history. Read once at start — a rotated token means a restart,
// which is cheaper than re-reading his credential every hour.
func newZenClient(tokenFile, baseURL string, timeout time.Duration) (*zenmoney.Client, error) {
	if tokenFile == "" {
		return nil, nil
	}
	raw, err := os.ReadFile(tokenFile)
	if err != nil {
		return nil, fmt.Errorf("read zenmoney token: %w", err)
	}
	return zenmoney.New(strings.TrimSpace(string(raw)), baseURL, timeout*zenTimeoutFactor)
}

// loop polls until the context is cancelled.
func (p *poller) loop(ctx context.Context, interval time.Duration) {
	p.pollOnce(ctx) // fire immediately; don't idle a full interval on start
	t := time.NewTicker(interval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			log.Printf("mavpoll: bye")
			return
		case <-t.C:
			p.pollOnce(ctx)
		}
	}
}

type poller struct {
	core       ipc.CoreAPI
	http       *http.Client
	netdataURL string
	kumaURL    string
	kumaKey    string
	wgIface    string
	wgCmd      string

	// kumaSeen — monitor name → state as of the last poll, so a monitor that
	// disappears from the gauge can be marked unknown instead of staying down.
	kumaSeen map[string]string

	// zen is nil unless a token file was configured — money tracking is a
	// capability, off by default like weather and telegram.
	zen      *zenmoney.Client
	zenEvery time.Duration
	zenLast  time.Time
}

// pollOnce — one sweep of every configured source. A failure in one logs and
// does NOT abort the rest: netdata being down shouldn't blind kuma.
func (p *poller) pollOnce(ctx context.Context) {
	now := time.Now()
	if p.netdataURL != "" {
		if err := p.pollNetdata(ctx, now); err != nil {
			log.Printf("mavpoll: netdata: %v", err)
		}
	}
	if p.kumaURL != "" {
		if err := p.pollKuma(ctx, now); err != nil {
			log.Printf("mavpoll: kuma: %v", err)
		}
	}
	if p.wgIface != "" {
		if err := p.pollWg(ctx); err != nil {
			log.Printf("mavpoll: wg: %v", err)
		}
	}
	// Money on its own, much slower cadence: a bank feed that updates hourly
	// polled every minute is 60 pointless reads of his financial history.
	if p.zen != nil && now.Sub(p.zenLast) >= p.zenEvery {
		p.zenLast = now
		if err := p.pollZenmoney(ctx, now); err != nil {
			log.Printf("mavpoll: zenmoney: %v", err)
		}
	}
}

// ---- zenmoney: spending/income totals → money facts ------------------------

// pollZenmoney reads today's and this month's totals and writes them as
// facts(kind=env, source=poll:zenmoney) (Vikunja #125).
//
// Two properties this function exists to hold:
//
//   - An empty or failed read writes NOTHING. zenmoney.Summary.Value() refuses
//     to encode a summary built from zero transactions, so a poller that cannot
//     reach the API leaves the last good fact in place rather than overwriting
//     it with a zero Maven would then recite as fact.
//   - Nothing about the money leaves the box except the diff request itself, to
//     the service that already holds his bank sessions. The totals are written
//     to the store and read back only when he asks; they are never search input
//     and no tick rule fires on them.
//
// Both windows are read from one diff call each. Two calls an hour against an
// API whose whole job is this is not worth caching.
//
// The write is UNCONDITIONAL, unlike every other poll in this file. The
// value-dedupe in writeIfChangedRaw only advances ts when the number moves, and
// for money that made ts mean "last changed" while the reader was asking it "as
// of when". A quiet 27 hours had core prefixing "данные от 30.07" to a figure
// that was current. The value now carries its own read stamp, so it differs
// every poll anyway and there is nothing left for the dedupe to catch.

// moneyWindow — one fact key and the period it covers.
type moneyWindow struct {
	key      string
	from, to time.Time
}

func (p *poller) pollZenmoney(ctx context.Context, now time.Time) error {
	dFrom, dTo := zenmoney.DayWindow(now)
	mFrom, mTo := zenmoney.MonthWindow(now)
	windows := []moneyWindow{
		{zenmoney.KeySpentToday, dFrom, dTo},
		{zenmoney.KeySpentMonth, mFrom, mTo},
	}
	var firstErr error
	for _, w := range windows {
		sum, err := p.zen.Since(ctx, w.from, w.to)
		if err != nil {
			if firstErr == nil {
				firstErr = err
			}
			continue
		}
		val, ok := sum.Value(now)
		if !ok {
			// Nothing read. Silence, not a zero. The last good fact stays, and
			// the window stamp inside it is what stops core reciting yesterday's
			// day total as today's after midnight.
			continue
		}
		if err := p.writeMoneyFact(ctx, w.key, val, now); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	return firstErr
}

// ---- wireguard: latest handshake → presence signal -------------------------

const (
	// wgFactKey / wgSource — the presence signal, read by the decay in core.
	// The source says infer because a handshake is evidence he is home, not a
	// reading of where he is.
	wgFactKey = "wg_handshake"
	wgSource  = "infer:wg"
)

// pollWg reads `wg show <iface> latest-handshakes` and writes a wg_handshake
// fact (source=infer:wg) stamped with the MOST RECENT peer handshake time — not
// now(). Presence decays from the real handshake instant, so the fact's ts must
// be that instant. We write only when the handshake ADVANCES vs the last fact,
// so a quiet tunnel produces no churn (and presence just decays out, τ=20min).
//
// `wg show` needs CAP_NET_ADMIN; run mavpoll with the cap or set -wg-cmd "sudo wg".
func (p *poller) pollWg(ctx context.Context) error {
	fields := strings.Fields(p.wgCmd)
	if len(fields) == 0 {
		return fmt.Errorf("wg command is empty")
	}
	args := append(fields[1:], "show", p.wgIface, "latest-handshakes")
	out, err := exec.CommandContext(ctx, fields[0], args...).Output()
	if err != nil {
		// wg says why it refused on stderr — usually a missing CAP_NET_ADMIN or
		// an interface that does not exist. Output() drops that, leaving a log
		// line that reads "exit status 1" and diagnoses nothing.
		var ee *exec.ExitError
		if errors.As(err, &ee) && len(ee.Stderr) > 0 {
			return fmt.Errorf("run %s: %w: %s", p.wgCmd, err, strings.TrimSpace(string(ee.Stderr)))
		}
		return fmt.Errorf("run %s: %w", p.wgCmd, err)
	}
	maxTs := parseMaxHandshake(string(out))
	if maxTs == 0 {
		return nil // no peer has ever handshaked → drop out of presence
	}
	hs := time.Unix(maxTs, 0)
	prev, err := p.core.LatestFactBySource(ctx, wgFactKey, wgSource)
	if err == nil && !hs.After(prev.Ts) {
		return nil // not newer → no churn
	}
	if err != nil && !isNoFact(err) {
		return fmt.Errorf("read %s: %w", wgFactKey, err)
	}
	// The ts is the handshake instant, not now(): presence decays from when he
	// was last seen.
	if err := p.writeFact(ctx, wgFactKey, wgSource, `"up"`, hs); err != nil {
		return err
	}
	log.Printf("mavpoll: %s @ %s (%s)", wgFactKey, hs.Format(time.RFC3339), wgSource)
	return nil
}

// parseMaxHandshake — max last-field unix ts across `wg show latest-handshakes`
// lines. Handles both the per-iface form (`<pubkey>\t<ts>`) and the `all` form
// (`<iface>\t<pubkey>\t<ts>`); the timestamp is always the last field. 0 = none.
func parseMaxHandshake(out string) int64 {
	var max int64
	for _, line := range strings.Split(out, "\n") {
		f := strings.Fields(line)
		if len(f) == 0 {
			continue
		}
		ts, err := strconv.ParseInt(f[len(f)-1], 10, 64)
		if err == nil && ts > max {
			max = ts
		}
	}
	return max
}

// ---- netdata: active alarms → aggregate severity ---------------------------

// netdata /api/v1/alarms?active=true returns {"alarms": {"<chart.name>": {...,
// "status": "WARNING"|"CRITICAL"|"CLEAR"|...}}}. We only need the max active
// severity; a rule fires on "critical". The per-alarm detail lives in netdata's
// own UI — we don't re-store it (YAGNI; add a per-alarm fact when a rule needs
// one specific alarm by name).
type netdataAlarms struct {
	Alarms map[string]struct {
		Status string `json:"status"`
	} `json:"alarms"`
}

func (p *poller) pollNetdata(ctx context.Context, now time.Time) error {
	body, err := p.get(ctx, p.netdataURL+"/api/v1/alarms?active=true", "")
	if err != nil {
		return err
	}
	var a netdataAlarms
	if err := json.Unmarshal(body, &a); err != nil {
		return fmt.Errorf("decode alarms: %w", err)
	}
	return p.writeIfChanged(ctx, "netdata_alarm", "poll:netdata", maxSeverity(a), now)
}

// maxSeverity reduces active alarms to the aggregate the rule consumes.
func maxSeverity(a netdataAlarms) string {
	sev := "clear"
	for _, al := range a.Alarms {
		switch strings.ToUpper(al.Status) {
		case "CRITICAL":
			return "critical" // highest — short-circuit
		case "WARNING":
			sev = "warning"
		}
	}
	return sev
}

// ---- kuma: monitor_status gauge → one fact per monitor ---------------------

// Kuma exposes Prometheus text: `monitor_status{...,monitor_name="X"} V` where
// V is 1=up 0=down 2=pending 3=maintenance. We write one fact per monitor,
// keyed `service_down:<monitor name>`, because the nudge has to say WHICH
// service is down. The aggregate this used to write could not, which is why
// the rule shipped disabled.
var (
	kumaLine = regexp.MustCompile(`^monitor_status\{([^}]*)\}\s+([0-9.eE+-]+)`)
	kumaName = regexp.MustCompile(`monitor_name="([^"]*)"`)
)

func (p *poller) pollKuma(ctx context.Context, now time.Time) error {
	body, err := p.get(ctx, p.kumaURL, p.kumaKey)
	if err != nil {
		return err
	}
	states := kumaMonitors(body)
	if len(states) == 0 {
		return fmt.Errorf("no monitor_status metrics (auth/endpoint wrong?)")
	}
	var firstErr error
	for name, val := range states {
		if err := p.writeIfChanged(ctx, kumaFactKey(name), kumaSource, val, now); err != nil && firstErr == nil {
			firstErr = err // one bad monitor must not blind the rest
		}
	}
	// A monitor deleted in kuma stops appearing in the gauge, and its last fact
	// would otherwise read "down" forever. Mark it unknown, which no rule fires
	// on. The seen-set is in memory, so a restart forgets it — harmless, since
	// the next poll that still lacks the monitor says nothing new either.
	for name := range p.kumaSeen {
		if _, still := states[name]; !still {
			if err := p.writeIfChanged(ctx, kumaFactKey(name), kumaSource, "unknown", now); err != nil && firstErr == nil {
				firstErr = err
			}
		}
	}
	p.kumaSeen = states
	return firstErr
}

// kumaSource — the provenance the loop rule requires. Written here, checked in
// loop.ServiceDownRule; a poller under any other source cannot fire it.
const kumaSource = "poll:uptimekuma"

// kumaFactKey — the fact key for one monitor. The suffix is the name he hears,
// so it stays as kuma spells it rather than being slugged into something else.
func kumaFactKey(name string) string { return "service_down:" + name }

// kumaMonitors parses kuma's Prometheus text into monitor name → state
// ("up"/"down"/"pending"/"maintenance"). An empty map means no monitor_status
// line matched at all (wrong endpoint, or auth rejected before the body).
// A line with no monitor_name label is skipped: a fact nobody can name is
// exactly the thing this replaced.
func kumaMonitors(body []byte) map[string]string {
	out := make(map[string]string)
	for _, line := range strings.Split(string(body), "\n") {
		m := kumaLine.FindStringSubmatch(strings.TrimSpace(line))
		if m == nil {
			continue
		}
		nm := kumaName.FindStringSubmatch(m[1])
		if nm == nil || strings.TrimSpace(nm[1]) == "" {
			continue
		}
		v, err := strconv.ParseFloat(m[2], 64)
		if err != nil {
			continue
		}
		out[strings.TrimSpace(nm[1])] = kumaState(v)
	}
	return out
}

// kumaState — the gauge's four values. pending and maintenance are not "down":
// a monitor paused in kuma should silence that monitor, not page him.
func kumaState(v float64) string {
	switch v {
	case 0:
		return "down"
	case 1:
		return "up"
	case 2:
		return "pending"
	case 3:
		return "maintenance"
	default:
		return "unknown"
	}
}

// ---- helpers ---------------------------------------------------------------

// factConfidence — every poll is a direct reading of another service, never an
// inference, so the fact goes in at full confidence.
const factConfidence = 1.0

// unchanged reports whether the latest fact for (key, source) already holds
// jsonVal. A missing fact is not an error here, it is the first write.
func (p *poller) unchanged(ctx context.Context, key, source, jsonVal string) (bool, error) {
	prev, err := p.core.LatestFactBySource(ctx, key, source)
	switch {
	case err == nil:
		return prev.Value == jsonVal, nil
	case isNoFact(err):
		return false, nil
	default:
		return false, fmt.Errorf("read %s: %w", key, err)
	}
}

// writeFact writes one `facts(kind=env)` row. Every poll in this file lands
// here, so the row shape is written once.
func (p *poller) writeFact(ctx context.Context, key, source, jsonVal string, now time.Time) error {
	_, err := p.core.WriteFact(ctx, ipc.WriteFactReq{
		Ts:         now,
		Kind:       "env",
		Key:        key,
		Value:      jsonVal,
		Source:     source,
		Confidence: factConfidence,
	})
	if err != nil {
		return fmt.Errorf("write %s: %w", key, err)
	}
	return nil
}

// writeIfChanged writes only when val differs from the latest fact for
// (key, source). Values are stored JSON-encoded (the store's convention:
// `"down"`, `"critical"`), matching how rules compare f.Value.
func (p *poller) writeIfChanged(ctx context.Context, key, source, val string, now time.Time) error {
	jv, _ := json.Marshal(val) // a string never fails to marshal
	same, err := p.unchanged(ctx, key, source, string(jv))
	if err != nil || same {
		return err // unchanged → no churn
	}
	if err := p.writeFact(ctx, key, source, string(jv), now); err != nil {
		return err
	}
	log.Printf("mavpoll: %s=%s (%s)", key, val, source)
	return nil
}

// writeIfChangedRaw is writeIfChanged for values that are already JSON (the
// money facts store an object, not a string). Kept separate rather than
// generalising writeIfChanged, because the string-valued env facts encoding
// their own value is the convention the rules rely on.
//
// The log line names the key and the source, never the figures: mavpoll's log
// is not the place his spending ends up.
func (p *poller) writeIfChangedRaw(ctx context.Context, key, source, jsonVal string, now time.Time) error {
	same, err := p.unchanged(ctx, key, source, jsonVal)
	if err != nil || same {
		return err
	}
	if err := p.writeFact(ctx, key, source, jsonVal, now); err != nil {
		return err
	}
	log.Printf("mavpoll: %s updated (%s)", key, source)
	return nil
}

// writeMoneyFact writes a money fact every poll, with no value comparison. See
// the comment above pollZenmoney for why this one does not go through
// writeIfChangedRaw.
//
// The log line names the key only, never the figures: mavpoll's log is not the
// place his spending ends up.
func (p *poller) writeMoneyFact(ctx context.Context, key, jsonVal string, now time.Time) error {
	if err := p.writeFact(ctx, key, zenmoney.Source, jsonVal, now); err != nil {
		return err
	}
	log.Printf("mavpoll: %s read (%s)", key, zenmoney.Source)
	return nil
}

// isNoFact — ErrNoFact rehydrated over the wire is wrapped (fmt.Errorf %w), so
// errors.Is is the right check.
func isNoFact(err error) bool {
	return errors.Is(err, ipc.ErrNoFact)
}

// maxBodyBytes caps what a source can make the poller hold. Kuma's whole
// metrics page is a few hundred kilobytes, so 4 MiB is slack, not a budget.
//
// Hitting the cap is an error, not a shorter body. A truncated kuma page parses
// cleanly right up to the cut, and every monitor past it reads as deleted — the
// poller would write "unknown" over live services and the down-rule would go
// quiet. Reading one byte past the cap is how we tell full from truncated.
const maxBodyBytes = 4 << 20

func (p *poller) get(ctx context.Context, url, basicUser string) ([]byte, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return nil, err
	}
	if basicUser != "" {
		req.SetBasicAuth("", basicUser) // kuma: API key as password, empty username
	}
	resp, err := p.http.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	body, err := io.ReadAll(io.LimitReader(resp.Body, maxBodyBytes+1))
	if err != nil {
		return nil, err
	}
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("GET %s: %s", url, resp.Status)
	}
	if len(body) > maxBodyBytes {
		return nil, fmt.Errorf("GET %s: body over %d bytes", url, maxBodyBytes)
	}
	return body, nil
}