homelab/ci/images/kuma-sync/internal/kumaclient/breml.go

package kumaclient

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

	kuma "github.com/breml/go-uptime-kuma-client"
	"github.com/breml/go-uptime-kuma-client/monitor"
	"github.com/breml/go-uptime-kuma-client/notification"
	"github.com/breml/go-uptime-kuma-client/tag"
)

const (
	ownerTagName  = "kuma-sync"
	ownerTagColor = "#2563eb"

	operationTimeout  = 30 * time.Second
	probeTimeout      = 10 * time.Second
	connectTimeout    = 30 * time.Second
	reconnectAttempts = 6
	reconnectMaxDelay = 8 * time.Second
)

// Breml is a Client backed by github.com/breml/go-uptime-kuma-client.
// It serialises all calls on one session. The socket.io transport does not
// reconnect, so when a call fails and a probe shows the session is dead, the
// session is rebuilt (session token first, password as fallback; bounded
// backoff) and the call is retried once. Tag ownership caveats: the kuma-sync
// tag is the lowest-ID tag of that name, and the library clobbers monitor
// conditions and retryOnlyOnStatusCodeFailure on update.
type Breml struct {
	mu    sync.Mutex
	c     *kuma.Client
	tagID int64

	sessCtx  context.Context
	url      string
	user     string
	pass     string
	opts     []kuma.Option
	token    string
	backoff0 time.Duration
}

var _ Client = (*Breml)(nil)

// NewBreml connects and logs in once, then ensures the kuma-sync tag exists.
// ctx bounds the lifetime of the socket session, not just the connect call.
//
// After a Kuma restart or dropped connection, the next call that fails with a
// dead session reconnects and retries once. Create is the exception: it is
// never retried, because its outcome can be unknown. Two windows can leave an
// untagged duplicate in Kuma, each signalled by an error wrapping
// ErrPossibleOrphan (the controller should log it at error level):
//  1. Kuma stored the monitor but the ack was lost.
//  2. Kuma stored the monitor, then tagging and the cleanup delete failed.
//
// In both cases the untagged monitor is never adopted or touched, so it needs
// manual cleanup in the Kuma UI, and the next reconcile may create a tagged
// monitor beside it.
func NewBreml(ctx context.Context, url, username, password string) (Client, error) {
	return newBreml(ctx, url, username, password)
}

func newBreml(ctx context.Context, url, username, password string, opts ...kuma.Option) (*Breml, error) {
	opts = append([]kuma.Option{kuma.WithOperationTimeout(operationTimeout)}, opts...)
	c, err := kuma.New(ctx, url, username, password, opts...)
	if err != nil {
		return nil, fmt.Errorf("connect to kuma: %w", err)
	}
	b := &Breml{
		c: c, sessCtx: ctx, url: url, user: username, pass: password, opts: opts,
		token: c.SessionToken(), backoff0: time.Second,
	}
	if b.tagID, err = b.ensureTag(ctx); err != nil {
		_ = c.Disconnect()
		return nil, err
	}
	return b, nil
}

// Close ends the session.
func (b *Breml) Close() error {
	return b.c.Disconnect()
}

// pickOwnerTag returns the lowest-ID tag named kuma-sync. Kuma does not
// enforce unique tag names.
func pickOwnerTag(tags []tag.Tag) (int64, bool) {
	var id int64
	found := false
	for _, t := range tags {
		if t.Name == ownerTagName && (!found || t.ID < id) {
			id, found = t.ID, true
		}
	}
	return id, found
}

// ownerKeyOf returns the kuma-sync tag value of a monitor. Different values on
// one monitor are a conflict, never resolved by picking one.
// The error is a *ConflictError listing all distinct values.
func ownerKeyOf(monitorID int64, tags []tag.MonitorTag, tagID int64) (string, error) {
	var keys []string
	for _, t := range tags {
		if t.TagID == tagID && t.Value != "" && !slices.Contains(keys, t.Value) {
			keys = append(keys, t.Value)
		}
	}
	switch len(keys) {
	case 0:
		return "", nil
	case 1:
		return keys[0], nil
	}
	slices.Sort(keys)
	return "", &ConflictError{MonitorID: monitorID, Keys: keys}
}

// createdDespite reports whether a CreateMonitor result means the monitor
// exists. The library returns the ID together with ErrUpdateEventTimeout when
// the monitor was created and only the update event is missing (library
// monitor.go: CreateMonitor doc and the errors.Is(err, ErrUpdateEventTimeout)
// branch).
func createdDespite(id int64, err error) bool {
	return err == nil || (id != 0 && errors.Is(err, kuma.ErrUpdateEventTimeout))
}

func (b *Breml) ensureTag(ctx context.Context) (int64, error) {
	tags, err := b.c.GetTags(ctx)
	if err != nil {
		return 0, err
	}
	if id, ok := pickOwnerTag(tags); ok {
		return id, nil
	}
	return b.c.CreateTag(ctx, tag.Tag{Name: ownerTagName, Color: ownerTagColor})
}

// recoverSession reports whether the session was rebuilt after err, so the
// caller may retry. Application-level errors and a probe on a healthy session
// return false.
func (b *Breml) recoverSession(ctx context.Context, err error) bool {
	rebuilt, _ := b.checkSession(ctx, err)
	return rebuilt
}

// checkSession probes the session after err. dead reports that the probe
// failed, i.e. err was a transport failure; rebuilt that it was then replaced.
func (b *Breml) checkSession(ctx context.Context, err error) (rebuilt, dead bool) {
	if ctx.Err() != nil {
		return false, false
	}
	for _, app := range []error{ErrNotFound, ErrNotOwned, ErrOwnerTagMissing, ErrOwnerConflict, ErrStatusPageNotFound} {
		if errors.Is(err, app) {
			return false, false
		}
	}
	pctx, cancel := context.WithTimeout(ctx, probeTimeout)
	defer cancel()
	if _, perr := b.c.GetTags(pctx); perr == nil {
		return false, false
	}
	return b.reconnect(ctx) == nil, true
}

func (b *Breml) reconnect(ctx context.Context) error {
	delay := b.backoff0
	var last error
	for attempt := 0; attempt < reconnectAttempts; attempt++ {
		if attempt > 0 {
			select {
			case <-time.After(delay):
			case <-ctx.Done():
				return ctx.Err()
			case <-b.sessCtx.Done():
				return b.sessCtx.Err()
			}
			delay = min(delay*2, reconnectMaxDelay)
		}
		opts := append(slices.Clone(b.opts), kuma.WithConnectTimeout(connectTimeout))
		if b.token != "" {
			opts = append(opts, kuma.WithSessionToken(b.token))
		}
		c, err := kuma.New(b.sessCtx, b.url, b.user, b.pass, opts...)
		if err != nil {
			last = err
			continue
		}
		_ = b.c.Disconnect()
		b.c = c
		if t := c.SessionToken(); t != "" {
			b.token = t
		}
		return nil
	}
	return fmt.Errorf("reconnect to kuma: %w", last)
}

// withRecovery runs op and, if it fails on a dead session, retries it once on
// a rebuilt one.
func withRecovery[T any](ctx context.Context, b *Breml, op func() (T, error)) (T, error) {
	v, err := op()
	if err != nil && b.recoverSession(ctx, err) {
		return op()
	}
	return v, err
}

func (b *Breml) ListOwned(ctx context.Context) ([]Monitor, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return withRecovery(ctx, b, func() ([]Monitor, error) { return b.listOwned(ctx) })
}

func (b *Breml) listOwned(ctx context.Context) ([]Monitor, error) {
	tags, err := b.c.GetTags(ctx)
	if err != nil {
		return nil, err
	}
	if !slices.ContainsFunc(tags, func(t tag.Tag) bool { return t.ID == b.tagID && t.Name == ownerTagName }) {
		return nil, ErrOwnerTagMissing
	}
	bases, err := b.c.GetMonitors(ctx)
	if err != nil {
		return nil, err
	}
	var out []Monitor
	var conflicts []error
	for _, base := range bases {
		key, err := ownerKeyOf(base.ID, base.Tags, b.tagID)
		if err != nil {
			conflicts = append(conflicts, err)
			continue
		}
		if key == "" {
			continue
		}
		m, err := toDomain(base, key)
		if err != nil {
			return nil, err
		}
		out = append(out, m.Normalized())
	}
	return out, errors.Join(conflicts...)
}

// Create is never retried. Two windows can leave an untagged monitor in Kuma,
// both signalled by an error wrapping ErrPossibleOrphan:
//  1. Kuma stored the monitor but the ack was lost (transport or operation
//     timeout error, no ID returned).
//  2. Kuma stored the monitor, then tagging failed and removing it failed too.
//
// Either way an untagged duplicate may exist; the controller never touches
// untagged monitors, so it needs manual cleanup in the Kuma UI. The next
// reconcile may create a tagged monitor beside it.
func (b *Breml) Create(ctx context.Context, m Monitor) (int64, error) {
	if m.OwnerKey == "" {
		return 0, errors.New("create: empty OwnerKey")
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	return flagUnknownCreate(
		func() (int64, error) { return b.createOnce(ctx, m) },
		func(err error) bool { _, dead := b.checkSession(ctx, err); return dead },
	)
}

// flagUnknownCreate runs create once. A failure without an ID that is a
// transport or operation-timeout error leaves the outcome unknown, so it is
// wrapped in ErrPossibleOrphan instead of being retried.
func flagUnknownCreate(create func() (int64, error), sessionDead func(error) bool) (int64, error) {
	id, err := create()
	if err == nil || id != 0 || errors.Is(err, ErrPossibleOrphan) {
		return id, err
	}
	if errors.Is(err, kuma.ErrOperationTimeout) || sessionDead(err) {
		return 0, errors.Join(err, fmt.Errorf("create outcome unknown: %w", ErrPossibleOrphan))
	}
	return 0, err
}

func (b *Breml) createOnce(ctx context.Context, m Monitor) (int64, error) {
	wire, err := toWire(m, nil)
	if err != nil {
		return 0, err
	}
	id, err := b.c.CreateMonitor(ctx, wire)
	if !createdDespite(id, err) {
		return 0, err
	}
	// Here the monitor exists, even if err is an update-event timeout.
	if _, terr := b.c.AddMonitorTag(ctx, b.tagID, id, m.OwnerKey); terr != nil {
		// An untagged leftover would be invisible to ListOwned.
		if derr := b.c.DeleteMonitor(ctx, id); derr != nil {
			return 0, errors.Join(terr, fmt.Errorf("remove untagged monitor %d: %w", id, derr),
				fmt.Errorf("monitor %d: %w", id, ErrPossibleOrphan))
		}
		return 0, terr
	}
	return id, nil
}

// lookupOwned returns the current monitor and verifies the owner tag.
func (b *Breml) lookupOwned(ctx context.Context, op string, id int64) (monitor.Base, string, error) {
	bases, err := b.c.GetMonitors(ctx)
	if err != nil {
		return monitor.Base{}, "", err
	}
	for _, base := range bases {
		if base.ID != id {
			continue
		}
		key, err := ownerKeyOf(base.ID, base.Tags, b.tagID)
		if err != nil {
			return monitor.Base{}, "", fmt.Errorf("%s %d: %w", op, id, err)
		}
		if key == "" {
			return monitor.Base{}, "", fmt.Errorf("%s %d: %w", op, id, ErrNotOwned)
		}
		return base, key, nil
	}
	return monitor.Base{}, "", fmt.Errorf("%s %d: %w", op, id, ErrNotFound)
}

type probeResult struct {
	found bool
	key   string
}

// Probe reports whether a monitor with this ID exists and which owner key it
// carries; an untagged monitor has "". Conflicting tag values return the
// *ConflictError.
func (b *Breml) Probe(ctx context.Context, id int64) (bool, string, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	res, err := withRecovery(ctx, b, func() (probeResult, error) {
		bases, err := b.c.GetMonitors(ctx)
		if err != nil {
			return probeResult{}, err
		}
		for _, base := range bases {
			if base.ID != id {
				continue
			}
			key, err := ownerKeyOf(base.ID, base.Tags, b.tagID)
			return probeResult{found: true, key: key}, err
		}
		return probeResult{}, nil
	})
	return res.found, res.key, err
}

func (b *Breml) Update(ctx context.Context, m Monitor) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	_, err := withRecovery(ctx, b, func() (struct{}, error) { return struct{}{}, b.update(ctx, m) })
	return err
}

func (b *Breml) update(ctx context.Context, m Monitor) error {
	cur, key, err := b.lookupOwned(ctx, "update", m.ID)
	if err != nil {
		return err
	}
	if key != m.OwnerKey {
		return fmt.Errorf("update %d: monitor carries owner key %q, not %q: %w", m.ID, key, m.OwnerKey, ErrNotOwned)
	}
	// Start from the full current object, also across a type change, so
	// fields we do not manage (proxy, resend interval, upside down, auth)
	// survive the overwrite.
	wire, err := toWire(m, &cur)
	if err != nil {
		return err
	}
	if err := b.c.UpdateMonitor(ctx, wire); err != nil {
		return err
	}
	// editMonitor does not change the active state.
	switch {
	case cur.IsActive && !m.Active:
		return b.c.PauseMonitor(ctx, m.ID)
	case !cur.IsActive && m.Active:
		return b.c.ResumeMonitor(ctx, m.ID)
	}
	return nil
}

func (b *Breml) Delete(ctx context.Context, id int64) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	_, err := withRecovery(ctx, b, func() (struct{}, error) {
		if _, _, err := b.lookupOwned(ctx, "delete", id); err != nil {
			return struct{}{}, err
		}
		return struct{}{}, b.c.DeleteMonitor(ctx, id)
	})
	return err
}

// DefaultNotifications reads the notification list cached by the session.
// Kuma has no command to request it: it pushes the list at login and to every
// session of the user after each change.
func (b *Breml) DefaultNotifications(ctx context.Context) ([]int64, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	return defaultIDs(b.c.GetNotifications(ctx)), nil
}

func defaultIDs(list []notification.Base) []int64 {
	var ids []int64
	for _, n := range list {
		if n.IsDefault {
			ids = append(ids, n.ID)
		}
	}
	slices.Sort(ids)
	return ids
}

// toDomain converts a monitor from the Kuma monitor list.
func toDomain(base monitor.Base, ownerKey string) (Monitor, error) {
	m := Monitor{
		ID:              base.ID,
		Type:            base.Type(),
		Name:            base.Name,
		Parent:          base.Parent,
		Active:          base.IsActive,
		Interval:        int32(base.Interval),
		RetryInterval:   int32(base.RetryInterval),
		MaxRetries:      int32(base.MaxRetries),
		NotificationIDs: slices.Clone(base.NotificationIDs),
		OwnerKey:        ownerKey,
	}
	if base.Description != nil {
		m.Description = *base.Description
	}
	switch m.Type {
	case "http":
		var v monitor.HTTP
		if err := base.As(&v); err != nil {
			return Monitor{}, err
		}
		if err := readHTTP(&m, v.HTTPDetails); err != nil {
			return Monitor{}, err
		}
	case "keyword":
		var v monitor.HTTPKeyword
		if err := base.As(&v); err != nil {
			return Monitor{}, err
		}
		if err := readHTTP(&m, v.HTTPDetails); err != nil {
			return Monitor{}, err
		}
		m.Keyword = v.Keyword
		m.InvertKeyword = v.InvertKeyword
	case "json-query":
		var v monitor.HTTPJSONQuery
		if err := base.As(&v); err != nil {
			return Monitor{}, err
		}
		if err := readHTTP(&m, v.HTTPDetails); err != nil {
			return Monitor{}, err
		}
		m.JSONPath = v.JSONPath
		m.JSONPathOperator = v.JSONPathOperator
		m.ExpectedValue = v.ExpectedValue
	case "dns":
		var v monitor.DNS
		if err := base.As(&v); err != nil {
			return Monitor{}, err
		}
		m.Hostname = v.Hostname
		m.Port = int32(v.Port)
		m.DNSResolveType = string(v.ResolveType)
		m.DNSResolveServer = v.ResolverServer
		for _, c := range v.Conditions {
			m.DNSConditions = append(m.DNSConditions, Condition{
				Variable: c.Variable, Operator: c.Operator, Value: c.Value, AndOr: string(c.AndOr),
			})
		}
	case "port":
		var v monitor.TCPPort
		if err := base.As(&v); err != nil {
			return Monitor{}, err
		}
		m.Hostname = v.Hostname
		m.Port = int32(v.Port)
	}
	return m, nil
}

func readHTTP(m *Monitor, d monitor.HTTPDetails) error {
	m.URL = d.URL
	m.Method = d.Method
	m.Body = d.Body
	m.AcceptedStatusCodes = slices.Clone(d.AcceptedStatusCodes)
	m.MaxRedirects = int32(d.MaxRedirects)
	m.IgnoreTLS = d.IgnoreTLS
	m.ExpiryNotification = d.ExpiryNotification
	m.Timeout = int32(math.Round(d.Timeout))
	if strings.TrimSpace(d.Headers) != "" {
		if err := json.Unmarshal([]byte(d.Headers), &m.Headers); err != nil {
			return fmt.Errorf("monitor %d: parse headers: %w", m.ID, err)
		}
	}
	return nil
}

// toWire builds the library monitor for m. With existing set, the typed
// object is first decoded from it so fields we do not manage are kept.
func toWire(m Monitor, existing *monitor.Base) (monitor.Monitor, error) {
	m = m.Normalized()
	load := func(target any) error {
		if existing == nil {
			return nil
		}
		return existing.As(target)
	}
	switch m.Type {
	case "group":
		v := &monitor.Group{}
		if err := load(v); err != nil {
			return nil, err
		}
		writeBase(&v.Base, m)
		return v, nil
	case "http":
		v := &monitor.HTTP{}
		if err := load(v); err != nil {
			return nil, err
		}
		writeBase(&v.Base, m)
		return v, writeHTTP(&v.HTTPDetails, m)
	case "keyword":
		v := &monitor.HTTPKeyword{}
		if err := load(v); err != nil {
			return nil, err
		}
		writeBase(&v.Base, m)
		v.Keyword = m.Keyword
		v.InvertKeyword = m.InvertKeyword
		return v, writeHTTP(&v.HTTPDetails, m)
	case "json-query":
		v := &monitor.HTTPJSONQuery{}
		if err := load(v); err != nil {
			return nil, err
		}
		writeBase(&v.Base, m)
		v.JSONPath = m.JSONPath
		v.JSONPathOperator = m.JSONPathOperator
		v.ExpectedValue = m.ExpectedValue
		return v, writeHTTP(&v.HTTPDetails, m)
	case "dns":
		v := &monitor.DNS{}
		if err := load(v); err != nil {
			return nil, err
		}
		writeBase(&v.Base, m)
		v.Hostname = m.Hostname
		v.Port = int(m.Port)
		v.ResolveType = monitor.DNSResolveType(m.DNSResolveType)
		v.ResolverServer = m.DNSResolveServer
		v.Conditions = nil
		for _, c := range m.DNSConditions {
			v.Conditions = append(v.Conditions, monitor.Condition{
				Variable: c.Variable, Operator: c.Operator, Value: c.Value, AndOr: monitor.ConditionOperator(c.AndOr),
			})
		}
		return v, nil
	case "port":
		v := &monitor.TCPPort{}
		if err := load(v); err != nil {
			return nil, err
		}
		writeBase(&v.Base, m)
		v.Hostname = m.Hostname
		v.Port = int(m.Port)
		return v, nil
	}
	return nil, fmt.Errorf("unsupported monitor type %q", m.Type)
}

func writeBase(b *monitor.Base, m Monitor) {
	b.ID = m.ID
	b.Name = m.Name
	desc := m.Description
	b.Description = &desc
	b.Parent = m.clone().Parent
	b.Interval = int64(m.Interval)
	b.RetryInterval = int64(m.RetryInterval)
	b.MaxRetries = int64(m.MaxRetries)
	b.IsActive = m.Active
	b.NotificationIDs = slices.Clone(m.NotificationIDs)
}

// writeHTTP expects m to be normalized.
func writeHTTP(d *monitor.HTTPDetails, m Monitor) error {
	d.URL = m.URL
	d.Timeout = float64(m.Timeout)
	d.ExpiryNotification = m.ExpiryNotification
	d.IgnoreTLS = m.IgnoreTLS
	d.MaxRedirects = int(m.MaxRedirects)
	d.AcceptedStatusCodes = slices.Clone(m.AcceptedStatusCodes)
	d.Method = m.Method
	d.Body = m.Body
	if d.HTTPBodyEncoding == "" {
		d.HTTPBodyEncoding = "json"
	}
	d.Headers = ""
	if len(m.Headers) > 0 {
		raw, err := json.Marshal(m.Headers)
		if err != nil {
			return fmt.Errorf("encode headers: %w", err)
		}
		d.Headers = string(raw)
	}
	return nil
}