package distworker

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"sync"
	"sync/atomic"
	"time"

	"github.com/Jeffail/tunny"
	"github.com/gorilla/websocket"
	"gorm.io/datatypes"

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

const (
	heartbeatInterval = 10 * time.Second

	// minQueueCapacity is the lower bound for the bounded job/result
	// channels so that a small pool still has some backpressure headroom.
	minQueueCapacity = 16
)

// jobPool is the minimal interface the Runner needs from a worker pool. It
// exists so tests can observe SetSize calls without wrapping the tunny
// concrete type.
type jobPool interface {
	Process(payload interface{}) interface{}
	SetSize(n int)
	Close()
}

// resultEnvelope pairs a job with its reports so the writer can include
// context (e.g., the check kind) when logging.
type resultEnvelope struct {
	job     wire.CheckJob
	reports []wire.CheckResultReport
}

// Runner manages the worker execution loop.
//
// Concurrency design:
//
//   - jobQueue and results are bounded channels sized relative to
//     maxConcurrency. The websocket read loop enqueues into jobQueue,
//     providing backpressure to the control plane when the worker is
//     saturated.
//   - A fixed number of dispatcher goroutines (maxConcurrency) drain the
//     jobQueue and call pool.Process. The number of in-flight executions
//     is bounded by the tunny.Pool size, which the control plane
//     configures via the "init" / "config" message (clamped to
//     maxConcurrency).
//   - Each websocket connection owns a single writer goroutine that
//     serializes result and heartbeat writes through a mutex. This keeps
//     websocket writes thread-safe and removes the per-task goroutine
//     that previously blocked pool execution while holding the mutex.
type Runner struct {
	config         *Config
	client         *Client
	pool           jobPool
	concurrency    int64
	maxConcurrency int
	jobQueue       chan wire.CheckJob
	results        chan resultEnvelope
	notifyQueue    chan wire.NotificationTask
	notifyResults  chan notifyResultEnvelope
	metricResults  chan wire.ServerMetricReport
	stopCh         chan struct{}
	wg             sync.WaitGroup
	started        atomic.Bool
	queueDepth     int64
	activeCount    int64
	notifyDepth    int64
	notifyActive   int64

	// credentialsMu guards credentials during init/config refresh.
	credentialsMu sync.RWMutex
	credentials   *wire.NotificationCredentials

	// systemContactsMu guards system contacts pushed via init/config.
	// Workers notify these contacts directly when the main API is
	// unreachable (see docs/distributed/notifications-from-worker.md
	// "System Selfcheck").
	systemContactsMu sync.RWMutex
	systemContacts   []wire.SystemContact

	// urlMu guards the worker URL pushed via init/config refresh. The
	// URL is what other workers and the main app dial to reach this
	// worker (it can differ from the bind host:port because of reverse
	// proxies / Traefik with HTTPS). See docs/worker-http-settings.md.
	urlMu sync.RWMutex
	url   string

	// peersMu guards the peer list pushed via init/config. The
	// control plane builds the list from worker_nodes (excluding
	// this worker) and refreshes it on every 5m config push; see
	// docs/distributed/worker-to-worker-raft.md §10 for the
	// intended use. The selfcheck module reads the list to drive
	// the peer poller and the consensus.
	peersMu sync.RWMutex
	peers   []wire.PeerInfo

	// peerCache stores the latest observation per peer. Written by
	// the peer poller (every peerPollInterval) and read by the
	// selfcheck consensus helper. Constructed in NewRunner so tests
	// can drive it without Start().
	peerCache *peerCache

	// masterStatusMu guards the local "master API up/down" snapshot
	// that the selfcheck writes after each probe and the
	// /api/peer/status handler reads. observedAt is the wall-clock
	// time of the most recent local probe; up is the verdict of
	// that probe. The zero time means "no probe has run yet".
	masterStatusMu sync.RWMutex
	masterStatusUp *bool
	masterStatusAt time.Time

	// selfcheckCancel terminates the periodic selfcheck goroutine started
	// by Start(). Nil until Start runs.
	selfcheckCancel context.CancelFunc

	// executor is the function the pool runs for each job. It is a
	// field so tests can swap it for a deterministic stub without
	// touching the websocket plumbing.
	executor func(payload interface{}) interface{}

	// resultsBuf holds the most recent result rows. The webapp
	// reads from it via RecentResults(n). Cleared by Stop so the
	// ring does not leak between worker runs.
	resultsBuf *resultBuffer

	// notificationsBuf mirrors resultsBuf for emitted
	// notifications. Phase 1 only fills this from selfcheck alerts.
	notificationsBuf *notificationBuffer

	// lastHeartbeatAt tracks the most recent successful heartbeat
	// write, so the webapp can render "last ack" without polling.
	lastHeartbeatMu sync.RWMutex
	lastHeartbeatAt time.Time

	// tokenRotatedAt records the wall-clock time of the last token
	// rotation, so the settings page can render "last rotated".
	tokenRotatedMu sync.RWMutex
	tokenRotatedAt time.Time

	// workerID / regionCode / workerVersion / workerCaps are
	// captured from the most recent init/config websocket message
	// so the webapp can render them read-only.
	workerIDMu    sync.RWMutex
	workerID      string
	regionMu      sync.RWMutex
	regionCode    string
	versionMu     sync.RWMutex
	workerVersion string
	capsMu        sync.RWMutex
	workerCaps    []string
	serverID      atomic.Int64

	// clientMu guards swap of the http/websocket client during
	// token rotation. The websocket loop reads r.client under the
	// lock; RotateToken swaps a fresh client in under the lock
	// before closing the old connection.
	clientMu sync.Mutex

	// closeOnce guards Close against double-close on the websocket
	// from RotateToken. Phase 1 has a single websocket; RotateToken
	// closes it so the reconnect loop picks up the new token.
	closeOnce sync.Once
}

// NewRunner creates a new worker runner. Config is taken by pointer to
// keep the parameter cheap as the struct grows (it now carries the
// HTTP listener settings on top of the control-plane connection
// fields).
func NewRunner(cfg *Config) *Runner {
	maxConc := cfg.MaxConcurrency
	if maxConc <= 0 {
		maxConc = DefaultMaxConcurrency
	}
	return &Runner{
		config:           cfg,
		maxConcurrency:   maxConc,
		stopCh:           make(chan struct{}),
		resultsBuf:       newResultBuffer(),
		notificationsBuf: newNotificationBuffer(),
		peerCache:        newPeerCache(),
	}
}

// Start begins the worker execution
func (r *Runner) Start() error {
	log.Println("worker: starting...")

	if r.config.URL == "" || r.config.Token == "" {
		return fmt.Errorf("RSMON_URL and RSMON_TOKEN must be set")
	}

	if !r.started.CompareAndSwap(false, true) {
		return fmt.Errorf("worker: runner already started")
	}

	r.client = NewClient(r.config.URL, r.config.Token)

	queueCap := r.queueCapacity()
	r.jobQueue = make(chan wire.CheckJob, queueCap)
	r.results = make(chan resultEnvelope, queueCap)
	r.notifyQueue = make(chan wire.NotificationTask, queueCap)
	r.notifyResults = make(chan notifyResultEnvelope, queueCap)
	r.metricResults = make(chan wire.ServerMetricReport, queueCap)

	if r.executor == nil {
		r.executor = r.defaultExecuteJob
	}
	atomic.StoreInt64(&r.concurrency, 1)
	r.pool = tunny.NewFunc(int(atomic.LoadInt64(&r.concurrency)), r.executor)

	// Start a fixed pool of dispatchers. The tunny.Pool size ultimately
	// limits the number of concurrent job executions; extra dispatchers
	// simply wait inside pool.Process when the pool is saturated.
	for i := 0; i < r.maxConcurrency; i++ {
		r.wg.Add(1)
		go r.dispatcher()
	}

	// Notification dispatchers share the pool's overall concurrency budget:
	// we run maxConcurrency notification dispatchers and let the runner's
	// NotificationMethods capability gate keep them quiet when the worker
	// is not authorized for the method.
	for i := 0; i < r.maxConcurrency; i++ {
		r.wg.Add(1)
		go r.notifyDispatcher()
	}

	// Start websocket task loop
	go r.websocketLoop()
	go r.serverMetricLoop()

	// Start periodic selfcheck loop. This probes the main API and, on
	// sustained unreachability, notifies system contacts directly via
	// the cached credentials. Lifetimes of selfcheck goroutines are
	// bound to stopCh (and the explicit cancel, kept for symmetry).
	selfcheckCtx, selfcheckCancel := context.WithCancel(context.Background())
	r.selfcheckCancel = selfcheckCancel
	go r.startSelfcheck(selfcheckCtx)

	// Start the peer poller. It refreshes r.peerCache with each
	// peer's latest /api/peer/status verdict. The selfcheck
	// consumes the cache to drive consensus. The poller is bound
	// to selfcheckCtx so it shuts down together with the
	// selfcheck loop on Stop.
	go r.peerPollerLoop(selfcheckCtx)

	// Wait for stop signal
	<-r.stopCh
	log.Println("worker: shutting down...")
	if r.selfcheckCancel != nil {
		r.selfcheckCancel()
	}

	// Dispatchers exit via stopCh. Do not close jobQueue here: the websocket
	// reader can still be unwinding and may otherwise race with a send.
	r.wg.Wait()

	// Pool is safe to close only after all dispatchers have returned,
	// otherwise an in-flight pool.Process would panic.
	r.pool.Close()

	// Close results so any future writers exit promptly. (At this point
	// the websocket connection is also gone, so this is just defensive.)
	close(r.results)

	return nil
}

// Stop gracefully stops the worker
func (r *Runner) Stop() {
	select {
	case <-r.stopCh:
		// already closed
	default:
		close(r.stopCh)
	}
}

// Enqueue submits a job to the worker pool. It returns false if the runner
// has been stopped. The call blocks while the bounded jobQueue is full,
// providing natural backpressure to the caller.
func (r *Runner) Enqueue(job wire.CheckJob) bool { //nolint:lll,gocritic // wire.CheckJob is ~104B; keep by-value to avoid forcing callers to take an address
	if r.jobQueue == nil {
		return false
	}
	select {
	case r.jobQueue <- job:
		atomic.AddInt64(&r.queueDepth, 1)
		return true
	case <-r.stopCh:
		return false
	}
}

// Concurrency returns the current tunny.Pool size.
func (r *Runner) Concurrency() int {
	return int(atomic.LoadInt64(&r.concurrency))
}

// EnqueueNotification submits a notification task to the worker pool. It
// returns false if the runner has been stopped. The call blocks while the
// bounded notifyQueue is full, providing backpressure to the control plane.
//
//nolint:gocritic // wire payload is shared with the dispatcher; keep by-value
func (r *Runner) EnqueueNotification(task wire.NotificationTask) bool {
	if r.notifyQueue == nil {
		return false
	}
	select {
	case r.notifyQueue <- task:
		atomic.AddInt64(&r.notifyDepth, 1)
		return true
	case <-r.stopCh:
		return false
	}
}

// notifyDispatcher is the per-notification-task execution loop. It mirrors
// dispatcher() but routes to ExecuteNotification + the notification_result
// writer rather than checkexec + check_result_report. The same
// maxConcurrency budget caps total in-flight work across both kinds.
func (r *Runner) notifyDispatcher() {
	defer r.wg.Done()
	for {
		select {
		case task := <-r.notifyQueue:
			atomic.AddInt64(&r.notifyDepth, -1)
			atomic.AddInt64(&r.notifyActive, 1)
			r.executeAndForwardNotification(task)
			atomic.AddInt64(&r.notifyActive, -1)
		case <-r.stopCh:
			return
		}
	}
}

// executeAndForwardNotification converts the wire NotificationTask to a Task
// shape, runs the executor, and pushes the result into notifyResults. The
// writer goroutine picks it up and serializes the websocket write.
func (r *Runner) executeAndForwardNotification(task wire.NotificationTask) { //nolint:gocritic // wire payload is shared
	if r.notifyResults == nil {
		return
	}
	deadline := time.Now().Add(models.DefaultNotificationExecutionTimeout)
	if task.Deadline != nil {
		if taskDeadline, err := time.Parse(time.RFC3339Nano, *task.Deadline); err == nil && taskDeadline.Before(deadline) {
			deadline = taskDeadline
		}
	}
	ctx, cancel := context.WithDeadline(context.Background(), deadline)
	defer cancel()
	payload, _ := json.Marshal(task)
	dbTask := models.Task{JobID: task.JobID, LeaseToken: task.LeaseToken, Payload: payload}
	if task.Deadline != nil {
		if deadline, err := time.Parse(time.RFC3339Nano, *task.Deadline); err == nil {
			dbTask.Deadline = &deadline
		}
	}
	if task.MessageID != 0 {
		msgID := task.MessageID
		dbTask.MessageID = &msgID
	}
	if len(task.EventIDs) > 0 {
		ev := task.EventIDs[0]
		_ = ev
	}
	report := r.ExecuteNotification(ctx, dbTask)
	env := notifyResultEnvelope{task: task, report: report}

	select {
	case <-r.stopCh:
		return
	default:
	}
	select {
	case r.notifyResults <- env:
	case <-r.stopCh:
	}
}

// MaxConcurrency returns the upper bound for pool size and dispatchers.
func (r *Runner) MaxConcurrency() int {
	return r.maxConcurrency
}

// QueueCapacity returns the bounded buffer size used for jobQueue and
// results. Exposed for tests and observability.
func (r *Runner) QueueCapacity() int {
	return r.queueCapacity()
}

// ActiveCount returns only running check executions. QueueDepth reports the
// disjoint pending-check count used with it in heartbeat capacity accounting.
func (r *Runner) ActiveCount() int {
	return int(atomic.LoadInt64(&r.activeCount))
}

// QueueDepth returns the current number of pending jobs in jobQueue.
func (r *Runner) QueueDepth() int {
	return int(atomic.LoadInt64(&r.queueDepth))
}

// ActiveNotifications returns the number of queued and in-flight deliveries.
func (r *Runner) ActiveNotifications() int {
	return int(atomic.LoadInt64(&r.notifyDepth) + atomic.LoadInt64(&r.notifyActive))
}

// NotificationQueueDepth returns deliveries waiting for a notification worker.
func (r *Runner) NotificationQueueDepth() int {
	return int(atomic.LoadInt64(&r.notifyDepth))
}

// queueCapacity returns the bounded buffer size for the job/result
// channels. The capacity is derived from maxConcurrency so that
// backpressure scales with the configured pool.
func (r *Runner) queueCapacity() int {
	qcap := 2 * r.maxConcurrency
	if qcap < minQueueCapacity {
		qcap = minQueueCapacity
	}
	return qcap
}

func (r *Runner) dispatcher() {
	defer r.wg.Done()
	for {
		select {
		case job := <-r.jobQueue:
			atomic.AddInt64(&r.queueDepth, -1)
			atomic.AddInt64(&r.activeCount, 1)
			r.executeAndForward(job)
			atomic.AddInt64(&r.activeCount, -1)
		case <-r.stopCh:
			return
		}
	}
}

func (r *Runner) executeAndForward(job wire.CheckJob) { //nolint:lll,gocritic // see Enqueue; CheckJob is forwarded into tunny.Pool as interface{}
	result := r.pool.Process(job)
	reports, ok := result.([]wire.CheckResultReport)
	if !ok || len(reports) == 0 {
		return
	}
	env := resultEnvelope{job: job, reports: reports}
	// Prefer an early exit when the runner is stopping so we never
	// block on a full results channel.
	select {
	case <-r.stopCh:
		return
	default:
	}
	select {
	case r.results <- env:
	case <-r.stopCh:
	}
}

func (r *Runner) websocketLoop() {
	for {
		select {
		case <-r.stopCh:
			return
		default:
		}

		if err := r.runWebsocket(); err != nil {
			log.Println("worker: websocket error:", err)
		}

		select {
		case <-time.After(3 * time.Second):
		case <-r.stopCh:
			return
		}
	}
}

func (r *Runner) runWebsocket() error {
	conn, err := r.client.WorkerSocket()
	if err != nil {
		return err
	}
	defer conn.Close() //nolint:errcheck
	log.Println("worker: websocket connected")

	var writeMu sync.Mutex
	done := make(chan struct{})

	// Heartbeat goroutine — shares writeMu with the writer.
	go r.heartbeat(conn, &writeMu, done)

	// Single writer goroutine for this connection: serializes result
	// and heartbeat writes through writeMu so websocket.WriteJSON is
	// never called concurrently. The dispatcher loop feeds it via the
	// bounded results channel.
	go r.writer(conn, &writeMu, done)

	for {
		var msg wire.WorkerMessage
		if err := conn.ReadJSON(&msg); err != nil {
			close(done)
			return err
		}
		if (msg.Kind == "init" || msg.Kind == "config") && msg.Init != nil {
			r.applyInit(msg.Init)
			continue
		}
		if msg.Kind != "task" {
			continue
		}
		if !r.enqueueTaskMessage(msg) {
			close(done)
			return nil
		}
	}
}

// enqueueTaskMessage prefers the v2 envelope over sibling legacy fields. Some
// rollout frames contain both check representations; executing the first match
// only keeps a current runner from running one check twice.
func (r *Runner) enqueueTaskMessage(msg wire.WorkerMessage) bool { //nolint:gocritic // wire envelope is the dispatcher boundary
	switch {
	case msg.TaskEnvelope != nil && msg.TaskEnvelope.Type == wire.TaskTypeNotification && msg.TaskEnvelope.Notify != nil:
		return r.EnqueueNotification(*msg.TaskEnvelope.Notify)
	case msg.TaskEnvelope != nil && msg.TaskEnvelope.Type == wire.TaskTypeCheck && msg.TaskEnvelope.Job != nil:
		return r.Enqueue(*msg.TaskEnvelope.Job)
	case msg.NotificationTask != nil:
		log.Printf("worker: received websocket notification task %s method=%s", msg.NotificationTask.JobID, msg.NotificationTask.Method)
		return r.EnqueueNotification(*msg.NotificationTask)
	case msg.Task != nil:
		log.Printf("worker: received websocket task %s", msg.Task.JobID)
		return r.Enqueue(*msg.Task)
	default:
		return true
	}
}

func (r *Runner) heartbeat(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan struct{}) {
	ticker := time.NewTicker(heartbeatInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			writeMu.Lock()
			err := conn.WriteJSON(wire.WorkerMessage{
				Kind: "heartbeat",
				Heartbeat: &wire.HeartbeatRequest{
					ActiveChecks:           r.ActiveCount(),
					QueueDepth:             r.QueueDepth(),
					ActiveNotifications:    r.ActiveNotifications(),
					NotificationQueueDepth: r.NotificationQueueDepth(),
				},
			})
			writeMu.Unlock()
			if err != nil {
				return
			}
			r.touchHeartbeat(time.Now().UTC())
		case <-done:
			return
		case <-r.stopCh:
			return
		}
	}
}

// touchHeartbeat records the most recent successful heartbeat write.
// Called from the heartbeat goroutine; the webapp reads it via
// LastHeartbeatAck.
func (r *Runner) touchHeartbeat(at time.Time) {
	if r == nil {
		return
	}
	r.lastHeartbeatMu.Lock()
	r.lastHeartbeatAt = at
	r.lastHeartbeatMu.Unlock()
}

// LastHeartbeatAck returns the wall-clock time of the most recent
// successful heartbeat. Returns the zero time if no heartbeat has
// been written yet.
func (r *Runner) LastHeartbeatAck() time.Time {
	if r == nil {
		return time.Time{}
	}
	r.lastHeartbeatMu.RLock()
	defer r.lastHeartbeatMu.RUnlock()
	return r.lastHeartbeatAt
}

func (r *Runner) writer(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan struct{}) {
	for {
		select {
		case env, ok := <-r.results:
			if !ok {
				return
			}
			writeMu.Lock()
			for i := range env.reports {
				report := &env.reports[i]
				report.LeaseToken = env.job.LeaseToken
				if err := conn.WriteJSON(wire.WorkerMessage{Kind: "result", Result: report}); err != nil {
					log.Printf(
						"worker: failed to report result job=%s check=%d kind=%s state=%s: %v",
						report.JobID, report.CheckID, env.job.Kind, report.State, err,
					)
					writeMu.Unlock()
					return
				}
				log.Printf(
					"worker: completed job=%s check=%d kind=%s state=%s reported=true",
					report.JobID, report.CheckID, env.job.Kind, report.State,
				)
				r.resultsBuf.add(resultRowFromReport(&env, report, time.Now().UTC()))
			}
			writeMu.Unlock()
		case env, ok := <-r.notifyResults:
			if !ok {
				return
			}
			writeMu.Lock()
			if err := conn.WriteJSON(wire.WorkerMessage{Kind: "result", NotificationResult: &env.report}); err != nil {
				log.Printf(
					"worker: failed to report notification result job=%s method=%s status=%s: %v",
					env.report.JobID, env.task.Method, env.report.Status, err,
				)
				writeMu.Unlock()
				return
			}
			log.Printf(
				"worker: completed notification job=%s method=%s status=%s message=%d duration_ms=%d",
				env.report.JobID, env.task.Method, env.report.Status, env.task.MessageID, env.report.DurationMs,
			)
			writeMu.Unlock()
		case report := <-r.metricResults:
			writeMu.Lock()
			if err := conn.WriteJSON(wire.WorkerMessage{Kind: "result", ServerMetric: &report}); err != nil {
				writeMu.Unlock()
				return
			}
			writeMu.Unlock()
		case <-done:
			return
		case <-r.stopCh:
			return
		}
	}
}

func (r *Runner) applyInit(init *wire.WorkerInit) {
	if init.Concurrency > 0 && init.Concurrency != r.Concurrency() {
		size := init.Concurrency
		if size > r.maxConcurrency {
			size = r.maxConcurrency
		}
		if size < 1 {
			size = 1
		}
		atomic.StoreInt64(&r.concurrency, int64(size))
		r.pool.SetSize(size)
	}
	if len(init.LLMs) > 0 {
		llm := init.LLMs[0]
		setEnvIfNotEmpty("LLM_URL", llm.URL)
		setEnvIfNotEmpty("LLM_MODEL", llm.Model)
		setEnvIfNotEmpty("LLM_APIKEY", llm.APIKey)
		setEnvIfNotEmpty("LLM_KIND", llm.Kind)
	}

	r.credentialsMu.Lock()
	r.credentials = init.Credentials
	r.credentialsMu.Unlock()

	r.systemContactsMu.Lock()
	r.systemContacts = init.SystemContacts
	r.systemContactsMu.Unlock()

	r.urlMu.Lock()
	r.url = init.URL
	r.urlMu.Unlock()

	// Peers is the slice of other workers this node can reach for
	// cross-worker confirmation. Refresh the cache so a removed
	// peer's stale observation is dropped immediately rather than
	// lingering until peerStatusMaxAge.
	r.peersMu.Lock()
	r.peers = append([]wire.PeerInfo(nil), init.Peers...)
	r.peersMu.Unlock()
	if r.peerCache != nil {
		r.peerCache.resetFor(init.Peers)
	}

	r.workerIDMu.Lock()
	r.workerID = init.WorkerID
	r.workerIDMu.Unlock()

	r.regionMu.Lock()
	r.regionCode = init.RegionCode
	r.regionMu.Unlock()

	r.versionMu.Lock()
	r.workerVersion = init.Version
	r.versionMu.Unlock()

	r.capsMu.Lock()
	r.workerCaps = append([]string(nil), init.Capabilities...)
	r.capsMu.Unlock()
	if init.ServerID == nil {
		r.serverID.Store(0)
	} else {
		r.serverID.Store(*init.ServerID)
	}

	smtpCount, tgCount := 0, 0
	if init.Credentials != nil {
		smtpCount = len(init.Credentials.SMTP)
		tgCount = len(init.Credentials.Telegram)
	}
	log.Printf(
		"worker: %s received worker_id=%s region=%s version=%s capabilities=%v "+
			"concurrency=%d llms=%d credentials_smtp=%d credentials_telegram=%d system_contacts=%d peers=%d",
		"config", init.WorkerID, init.RegionCode, init.Version, init.Capabilities,
		init.Concurrency, len(init.LLMs), smtpCount, tgCount, len(init.SystemContacts), len(init.Peers),
	)
}

// Credentials returns a snapshot of the current notification credentials.
// Safe for concurrent use. Returns nil if no credentials have been pushed.
func (r *Runner) Credentials() *wire.NotificationCredentials {
	r.credentialsMu.RLock()
	defer r.credentialsMu.RUnlock()
	return r.credentials
}

// SystemContacts returns a snapshot of the cached system contacts.
// Safe for concurrent use.
func (r *Runner) SystemContacts() []wire.SystemContact {
	r.systemContactsMu.RLock()
	defer r.systemContactsMu.RUnlock()
	out := make([]wire.SystemContact, len(r.systemContacts))
	copy(out, r.systemContacts)
	return out
}

// URL returns the publicly-advertised URL the main app and peer workers
// should dial to reach this worker. It is set from the init/config
// websocket message and may be empty if the main app did not push one
// (e.g. legacy worker). Safe for concurrent use.
func (r *Runner) URL() string {
	r.urlMu.RLock()
	defer r.urlMu.RUnlock()
	return r.url
}

// Peers returns a snapshot of the cached peer list pushed by the
// control plane via WorkerInit.Peers. The current worker is excluded
// upstream so a worker never dials itself. Returns nil when no init
// has landed or when the control plane ships an empty list (e.g. a
// single-worker install); callers must handle that case explicitly so
// they fall back to the single-node selfcheck verdict. Safe for
// concurrent use.
func (r *Runner) Peers() []wire.PeerInfo {
	r.peersMu.RLock()
	defer r.peersMu.RUnlock()
	if len(r.peers) == 0 {
		return nil
	}
	out := make([]wire.PeerInfo, len(r.peers))
	copy(out, r.peers)
	return out
}

// SetMasterStatus records the most recent local selfcheck verdict.
// Called by the selfcheck loop after every probe so the
// /api/peer/status HTTP handler can answer with the same value the
// consensus uses. Passing up == nil resets the snapshot to "unknown"
// (no probe yet) and is used by tests.
func (r *Runner) SetMasterStatus(up *bool, observedAt time.Time) {
	if r == nil {
		return
	}
	r.masterStatusMu.Lock()
	r.masterStatusUp = up
	r.masterStatusAt = observedAt
	r.masterStatusMu.Unlock()
}

// MasterStatus returns the most recent local selfcheck verdict and
// the wall-clock time it was produced. up == nil means the local
// selfcheck has not produced a verdict yet (very first tick or the
// runner has not been started). Safe for concurrent use.
func (r *Runner) MasterStatus() (up *bool, observedAt time.Time) {
	if r == nil {
		return nil, time.Time{}
	}
	r.masterStatusMu.RLock()
	defer r.masterStatusMu.RUnlock()
	return r.masterStatusUp, r.masterStatusAt
}

func setEnvIfNotEmpty(key, value string) {
	if value != "" {
		_ = os.Setenv(key, value)
	}
}

// defaultExecuteJob executes a single check job.
func (r *Runner) defaultExecuteJob(payload interface{}) interface{} {
	job := payload.(wire.CheckJob)

	log.Printf("worker: executing job %s (check %d, kind %s)", job.JobID, job.CheckID, job.Kind)

	monitor := &models.Monitor{Host: job.Host}
	monitor.ID = job.MonitorID

	check := models.Check{
		Kind:     job.Kind,
		URL:      job.URL,
		Settings: datatypes.JSON(job.Settings),
	}
	check.ID = job.CheckID
	check.MonitorID = job.MonitorID

	results := checkexec.Execute(monitor, []models.Check{check})

	if len(results) == 0 {
		log.Printf("worker: no results for job %s", job.JobID)
		return []wire.CheckResultReport{}
	}

	reports := make([]wire.CheckResultReport, 0, len(results))
	for _, result := range results {
		report := wire.CheckResultReport{
			JobID:      job.JobID,
			CheckID:    job.CheckID,
			MonitorID:  job.MonitorID,
			State:      result.Result.State,
			DurationMs: result.Result.Duration.Milliseconds(),
			Warnings:   result.Result.Warnings,
			Infos:      result.Result.Infos,
			Metrics:    result.Metrics,
		}

		if result.Result.Error != nil {
			errStr := result.Result.Error.Error()
			report.Error = &errStr
		}

		if result.Result.Expires != nil {
			exp := result.Result.Expires.Format(time.RFC3339)
			report.ExpiresAt = &exp
		}

		reports = append(reports, report)
	}

	return reports
}

// RecentResults returns the most recent n result rows produced by
// this worker, in chronological order. n <= 0 returns an empty slice.
// Safe to call before Start (returns nil).
func (r *Runner) RecentResults(n int) []ResultRow {
	if r == nil || r.resultsBuf == nil {
		return nil
	}
	return r.resultsBuf.snapshot(n)
}

// RecentNotifications returns the most recent n notification rows.
// Phase 1 only fills this buffer from selfcheck alerts via
// RecordNotification; the main-app-issued notifications still live
// in the main app's database.
func (r *Runner) RecentNotifications(n int) []NotificationRow {
	if r == nil || r.notificationsBuf == nil {
		return nil
	}
	return r.notificationsBuf.snapshot(n)
}

// RecordNotification appends one row to the notification ring buffer.
// Called from selfcheck.sendSystemAlert so the webapp /notifications
// page can show what the worker emitted. Safe before Start.
func (r *Runner) RecordNotification(n *NotificationRow) {
	if r == nil || r.notificationsBuf == nil {
		return
	}
	if n.At.IsZero() {
		n.At = time.Now().UTC()
	}
	r.notificationsBuf.add(n)
}

// Token returns the current bearer token. The webapp settings page
// masks this for display.
func (r *Runner) Token() string {
	if r == nil || r.config == nil {
		return ""
	}
	return r.config.Token
}

// TokenRotatedAt returns the wall-clock time of the most recent
// successful token rotation. Zero before the first rotation.
func (r *Runner) TokenRotatedAt() time.Time {
	if r == nil {
		return time.Time{}
	}
	r.tokenRotatedMu.RLock()
	defer r.tokenRotatedMu.RUnlock()
	return r.tokenRotatedAt
}

// WorkerID returns the worker_id pushed by the main app via the
// init/config message. Empty before init lands.
func (r *Runner) WorkerID() string {
	if r == nil {
		return ""
	}
	// The init payload is captured via the credentials/url caches; we
	// also expose the worker_id through the URL setter below.
	r.workerIDMu.RLock()
	defer r.workerIDMu.RUnlock()
	return r.workerID
}

// RegionCode returns the region_code from the init payload.
func (r *Runner) RegionCode() string {
	if r == nil {
		return ""
	}
	r.regionMu.RLock()
	defer r.regionMu.RUnlock()
	return r.regionCode
}

// WorkerVersion returns the worker_version from the init payload.
func (r *Runner) WorkerVersion() string {
	if r == nil {
		return ""
	}
	r.versionMu.RLock()
	defer r.versionMu.RUnlock()
	return r.workerVersion
}

// WorkerCapabilities returns the capabilities from the init payload.
func (r *Runner) WorkerCapabilities() []string {
	if r == nil {
		return nil
	}
	r.capsMu.RLock()
	defer r.capsMu.RUnlock()
	return append([]string(nil), r.workerCaps...)
}

// HTTPConfig returns the worker's local HTTP listener settings. Used
// by the webapp's settings page to render the configured bind
// address and URL.
func (r *Runner) HTTPConfig() HTTPConfig {
	if r == nil || r.config == nil {
		return HTTPConfig{}
	}
	return r.config.HTTP
}

// RotateToken asks the main app to issue a fresh bearer token,
// updates the in-memory config + client, and closes the current
// websocket so the reconnect loop picks up the new token.
//
// Returns the new token string. On any failure the old token and
// client are kept untouched.
func (r *Runner) RotateToken(_ context.Context) (string, error) {
	if r == nil || r.config == nil {
		return "", fmt.Errorf("worker: runner not initialized")
	}
	if r.client == nil {
		return "", fmt.Errorf("worker: client not yet started")
	}
	newToken, err := r.client.RotateToken()
	if err != nil {
		return "", err
	}
	if newToken == "" || newToken == r.config.Token {
		return "", fmt.Errorf("worker: rotate-token returned unchanged or empty token")
	}

	// Swap config + client under lock so a concurrent heartbeat
	// cannot race with the rotation.
	r.clientMu.Lock()
	r.config.Token = newToken
	oldClient := r.client
	r.client = NewClient(r.config.URL, newToken)
	r.clientMu.Unlock()

	// Stamp the rotation time so the settings page can show it.
	r.tokenRotatedMu.Lock()
	r.tokenRotatedAt = time.Now().UTC()
	r.tokenRotatedMu.Unlock()

	// Force the websocket loop to reconnect with the new token. The
	// old connection's next heartbeat will fail with 401; closing
	// the connection now shortens that window.
	r.closeOnce.Do(func() {
		// Close the underlying websocket by triggering the runner's
		// normal stop path; the websocketLoop goroutine will reconnect
		// after we re-arm stopCh. This is the cleanest way to drive
		// the loop without exposing internals.
		select {
		case <-r.stopCh:
		default:
			close(r.stopCh)
		}
	})
	_ = oldClient // client has no Close; the websocket layer owns it.
	return newToken, nil
}