worker/internal/distworker/selfcheck.go

package distworker

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	mathrand "math/rand/v2"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"time"

	"rocketgit.ru/rsmon/worker/app/models"
	"rocketgit.ru/rsmon/worker/app/models/concerns"
	"rocketgit.ru/rsmon/worker/internal/checkexec"
	"rocketgit.ru/rsmon/worker/internal/notify"
	"rocketgit.ru/rsmon/worker/internal/wire"
)

const (
	selfcheckInterval      = 30 * time.Second
	selfcheckTimeoutMillis = 10_000
	selfcheckJitterMax     = 30 * time.Second
	selfcheckProbePath     = "/up"
	// selfcheckConsensusWait is the time the cluster-level down
	// verdict must hold before the selfcheck fires a system alert.
	// 5 minutes matches the user-facing requirement: a single
	// blip should not page, but a sustained outage must.
	selfcheckConsensusWait   = 5 * time.Minute
	selfcheckDownMessage     = "Нет связи с основным api"
	selfcheckRecoveryMessage = "Связь с основным api восстановлена"
	selfcheckLeaderMessage   = "Изменился мастер воркер оповещений"
	selfcheckStateOK         = "OK"
	selfcheckStateWarn       = "WARN"
	// systemContactKindEmail is the wire-level identifier for email
	// system contacts, matching Contact.Kind values from the control plane.
	systemContactKindEmail = "email"

	// notificationChannel labels are short tags the webapp renders
	// alongside each row on /notifications. Kept distinct from the
	// systemContactKind* constants so a single contact can be
	// reached over multiple channels without an extra column.
	notificationChannelSMTP     = "smtp"
	notificationChannelTelegram = "telegram"
)

// selfcheckState tracks the cluster-level up/down verdict for the
// master API as decided by simple-majority consensus over self +
// reachable peers. Alert fires only after the down verdict has held
// for selfcheckConsensusWait; recovery fires only after a matching
// up verdict has held for the same window.
type selfcheckState struct {
	consensus *consensusState
}

// startSelfcheck launches the periodic selfcheck loop. The first tick is
// delayed by a random 0..selfcheckJitterMax so a fleet restart does not
// stampede the main API simultaneously. Returns when ctx is canceled.
func (r *Runner) startSelfcheck(ctx context.Context) {
	jitter := time.Duration(mathrand.Int64N(int64(selfcheckJitterMax))) //nolint:gosec // non-cryptographic jitter to stagger fleet probes
	select {
	case <-time.After(jitter):
	case <-ctx.Done():
		return
	}

	state := &selfcheckState{consensus: &consensusState{}}
	ticker := time.NewTicker(selfcheckInterval)
	defer ticker.Stop()

	r.runSelfcheckOnce(ctx, state)

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			r.runSelfcheckOnce(ctx, state)
		}
	}
}

func (r *Runner) runSelfcheckOnce(ctx context.Context, state *selfcheckState) {
	target := selfcheckTarget(r.config.URL)
	if target == "" {
		return
	}

	now := time.Now()
	ok := runMainAPIHTTPCheck(target)
	upFlag := ok
	r.SetMasterStatus(&upFlag, now)

	peers := r.peerObservationsForConsensus(now)
	configuredPeers := len(r.Peers())
	// In a multi-worker deployment we need a peer-backed quorum
	// (self + at least one fresh peer) before any verdict is
	// reported. A lone self vote — peers configured but no fresh
	// peer observations yet — is consensusNoQuorum so the
	// selfcheck state machine does not start its 5-minute down
	// timer on what might be a transient fleet-bootstrap blip.
	// The configured peer list itself is the signal that the
	// operator wants cross-worker consensus; peerObservations
	// above already drops stale/error observations so "configured
	// peers" maps 1:1 to "peer observations can be fresh".
	minVotes := 1
	if configuredPeers > 0 {
		minVotes = 2
	}
	verdict := decideConsensus(&upFlag, peers, minVotes)
	up, down, total := tallyConsensus(&upFlag, peers)
	voters := consensusVoters(r.WorkerID(), peers)
	notificationLeader := consensusNotificationLeader(voters)
	log.Printf("worker: selfcheck verdict=%s up=%d down=%d total=%d min_votes=%d target=%s",
		verdict, up, down, total, minVotes, target)
	if oldLeader, changed := state.consensus.notificationLeaderChanged(notificationLeader); changed && r.isConsensusNotificationLeader(notificationLeader) {
		log.Printf("worker: selfcheck notification leader changed old=%s new=%s; firing system alert", oldLeader, notificationLeader)
		r.sendSystemAlert(ctx, true, r.formatSelfcheckLeaderMessage(oldLeader, target, verdict, up, down, total, minVotes, configuredPeers, len(peers), notificationLeader, voters))
	}

	if verdict == consensusDown {
		if state.consensus.isDownConsensusHeld(verdict, now) &&
			!state.consensus.alertActive {
			state.consensus.markDownAlertFired()
			if !r.isConsensusNotificationLeader(notificationLeader) {
				log.Printf("worker: selfcheck down consensus held %s; notification leader=%s; skipping send", selfcheckConsensusWait, notificationLeader)
				return
			}
			log.Printf("worker: selfcheck down consensus held %s; firing system alert", selfcheckConsensusWait)
			r.sendSystemAlert(ctx, true, r.formatSelfcheckMessage(selfcheckDownMessage, target, verdict, up, down, total, minVotes, configuredPeers, len(peers), notificationLeader, voters))
		}
		return
	}

	if verdict == consensusUp && state.consensus.shouldFireRecovery(verdict, now) {
		state.consensus.markRecoveryFired()
		if !r.isConsensusNotificationLeader(notificationLeader) {
			log.Printf("worker: selfcheck up consensus held %s; notification leader=%s; skipping recovery", selfcheckConsensusWait, notificationLeader)
			return
		}
		log.Printf("worker: selfcheck up consensus held %s; firing recovery", selfcheckConsensusWait)
		r.sendSystemAlert(ctx, false, r.formatSelfcheckMessage(selfcheckRecoveryMessage, target, verdict, up, down, total, minVotes, configuredPeers, len(peers), notificationLeader, voters))
	}
}

func (r *Runner) formatSelfcheckLeaderMessage(
	oldLeader, target string,
	verdict consensusDecision,
	up, down, total, minVotes, configuredPeers, freshPeers int,
	notificationLeader string,
	voters []string,
) string {
	return r.formatSelfcheckMessage(
		fmt.Sprintf("%s: %s -> %s", selfcheckLeaderMessage, oldLeader, notificationLeader),
		target, verdict, up, down, total, minVotes, configuredPeers, freshPeers, notificationLeader, voters,
	)
}

func (r *Runner) formatSelfcheckMessage(
	message, target string,
	verdict consensusDecision,
	up, down, total, minVotes, configuredPeers, freshPeers int,
	notificationLeader string,
	voters []string,
) string {
	workerID := r.WorkerID()
	if workerID == "" {
		workerID = "unknown"
	}
	region := r.RegionCode()
	if region == "" {
		region = "unknown"
	}
	return fmt.Sprintf(
		"%s\n\nWorker: %s\nRegion: %s\nTarget: %s\nRaft status: lightweight peer quorum\nNotification leader: %s\nConsensus: %s\nVotes: up=%d down=%d total=%d min_votes=%d\nWorkers seen: %d current votes, %d fresh peers of %d configured peers\nVoters: %s\nHold time: %s",
		message, workerID, region, target, notificationLeader, verdict, up, down, total, minVotes, total, freshPeers, configuredPeers, strings.Join(voters, ","), selfcheckConsensusWait,
	)
}

func (r *Runner) isConsensusNotificationLeader(leader string) bool {
	return leader != "" && r.WorkerID() == leader
}

func consensusVoters(self string, peers []peerObservation) []string {
	voters := make([]string, 0, len(peers)+1)
	if self != "" {
		voters = append(voters, self)
	}
	for _, peer := range peers {
		if peer.WorkerID != "" {
			voters = append(voters, peer.WorkerID)
		}
	}
	sort.Strings(voters)
	return voters
}

func consensusNotificationLeader(voters []string) string {
	if len(voters) == 0 {
		return ""
	}
	return voters[0]
}

func selfcheckTarget(baseURL string) string {
	baseURL = strings.TrimRight(baseURL, "/")
	if baseURL == "" {
		return ""
	}
	return baseURL + selfcheckProbePath
}

func runMainAPIHTTPCheck(target string) bool {
	settings, _ := json.Marshal(models.CheckSettings{
		ExpectedAnswer: "default",
		RequestMethod:  "GET",
		Timeout:        selfcheckTimeoutMillis,
		SlowTime:       selfcheckTimeoutMillis,
	})
	name := "main api selfcheck"
	monitor := &models.Monitor{Host: selfcheckHost(target)}
	check := models.Check{
		ID:       -1,
		Name:     &name,
		Kind:     "http", //nolint:goconst // check kind is a wire string, not the http package identifier
		Interval: int(selfcheckInterval.Seconds()),
		URL:      &target,
		Settings: settings,
	}
	results := checkexec.Execute(monitor, []models.Check{check})
	if len(results) == 0 {
		return false
	}
	state := results[0].Result.State
	return state == selfcheckStateOK || state == selfcheckStateWarn
}

func selfcheckHost(target string) string {
	parsed, err := url.Parse(target)
	if err != nil || parsed.Host == "" {
		return target
	}
	return parsed.Host
}

// sendSystemAlert notifies all cached system contacts using cached credentials.
// failure=true means "down" message, false means "recovered".
func (r *Runner) sendSystemAlert(_ context.Context, failure bool, message string) {
	creds := r.Credentials()
	if creds == nil {
		log.Printf("worker: selfcheck alert skipped: no credentials")
		return
	}

	contacts := r.SystemContacts()
	if len(contacts) == 0 {
		log.Printf("worker: selfcheck alert skipped: no system contacts")
		return
	}

	subject := "RSMon worker alert"
	if !failure {
		subject = "RSMon worker recovery"
	}

	for i := range contacts {
		c := contacts[i]
		switch c.Kind {
		case systemContactKindEmail:
			if len(creds.SMTP) == 0 {
				continue
			}
			cred := smtpCredToModel(&creds.SMTP[0])
			now := time.Now().UTC()
			row := &NotificationRow{
				Kind:    c.Kind,
				Channel: notificationChannelSMTP,
				Subject: subject,
				Body:    message,
				At:      now,
			}
			if err := notify.Email(cred, c.Value, subject, message, ""); err != nil {
				log.Printf("worker: selfcheck email to %s failed: %v", c.Value, err)
				row.OK = false
				row.Error = err.Error()
			} else {
				row.OK = true
			}
			r.RecordNotification(row)
		case "telegram_private", "telegram_group":
			if len(creds.Telegram) == 0 {
				continue
			}
			chatID, err := parseTelegramChatID(c.Value)
			if err != nil {
				log.Printf("worker: selfcheck telegram chat_id parse failed for %s: %v", c.Value, err)
				continue
			}
			cred := telegramCredToModel(creds.Telegram[0])
			body := fmt.Sprintf("%s\n\n%s", subject, message)
			now := time.Now().UTC()
			row := &NotificationRow{
				Kind:    c.Kind,
				Channel: notificationChannelTelegram,
				Subject: subject,
				Body:    message,
				At:      now,
			}
			if err := notify.Telegram(cred, chatID, body); err != nil {
				log.Printf("worker: selfcheck telegram to %s failed: %v", c.Value, err)
				row.OK = false
				row.Error = err.Error()
			} else {
				row.OK = true
			}
			r.RecordNotification(row)
		}
	}
}

// parseTelegramChatID converts a numeric telegram chat id stored as a
// string into an int64.
func parseTelegramChatID(raw string) (int64, error) {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return 0, fmt.Errorf("empty chat id")
	}
	id, err := strconv.ParseInt(raw, 10, 64)
	if err != nil {
		return 0, fmt.Errorf("parse chat id: %w", err)
	}
	return id, nil
}

// smtpCredToModel wraps a wire SMTP credential in a models.NotificationCredential.
// SecretEnc carries the plaintext with the "plain:" prefix so models.GetSecret
// returns it unchanged — workers do not have the encryption key configured.
func smtpCredToModel(c *wire.SMTPCredential) *models.NotificationCredential {
	port := c.Port
	enabled := true
	return &models.NotificationCredential{
		Model:              concerns.Model{ID: c.ID},
		Kind:               models.CredentialKindSMTP,
		Name:               c.Name,
		Server:             &c.Server,
		Port:               &port,
		Login:              &c.Login,
		FromName:           &c.FromName,
		FromAddr:           &c.FromAddress,
		InsecureSkipVerify: c.InsecureSkipVerify,
		Enabled:            &enabled,
		SecretEnc:          "plain:" + c.Password,
	}
}

// telegramCredToModel wraps a wire Telegram credential in a
// models.NotificationCredential. SecretEnc carries the plaintext with the
// "plain:" prefix so models.GetSecret returns it unchanged.
func telegramCredToModel(c wire.TelegramCredential) *models.NotificationCredential {
	enabled := true
	cred := &models.NotificationCredential{
		Model:     concerns.Model{ID: c.ID},
		Kind:      models.CredentialKindTelegram,
		Name:      c.Name,
		BotName:   &c.BotName,
		APIURL:    &c.APIURL,
		Enabled:   &enabled,
		SecretEnc: "plain:" + c.Token,
	}
	return cred
}