worker/app/models/network_diagnostics.go

package models

import (
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/google/uuid"
	"gorm.io/datatypes"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"

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

const (
	ConfirmStateNone      = "none"
	ConfirmStatePending   = "pending"
	ConfirmStateConfirmed = "confirmed" // A different worker reproduced the failure.
	ConfirmStateTimeout   = "confirmed_by_timeout"
	AttemptKindRegular    = "regular"
	AttemptKindConfirm    = "confirmation"
	AttemptStateQueued    = "queued"
	AttemptStateLeased    = "leased"
	AttemptStateFinished  = "finished"
	diagnosticSoft        = "soft"
	diagnosticHard        = "hard"
	diagnosticRecovery    = "recovery"
)

type diagnosticSettings struct {
	confirmTimeout time.Duration
	healthWindow   time.Duration
	healthRate     float64
	healthMin      int64
}

func settingsForAccount(account *Account) diagnosticSettings {
	settings := diagnosticSettings{confirmTimeout: 90 * time.Second, healthWindow: 5 * time.Minute, healthRate: .5, healthMin: 10}
	if account == nil || account.Plan == nil {
		return settings
	}
	plan := account.Plan
	if plan.ConfirmTimeoutSec > 0 {
		settings.confirmTimeout = time.Duration(plan.ConfirmTimeoutSec) * time.Second
	}
	if plan.HealthWindowSec > 0 {
		settings.healthWindow = time.Duration(plan.HealthWindowSec) * time.Second
	}
	if plan.HealthRateThreshold > 0 && plan.HealthRateThreshold <= 1 {
		settings.healthRate = plan.HealthRateThreshold
	}
	if plan.HealthMinAttempts > 0 {
		settings.healthMin = int64(plan.HealthMinAttempts)
	}
	if !plan.Confirmations {
		return settings
	}
	if account.ConfirmTimeoutSec != nil && *account.ConfirmTimeoutSec >= 15 {
		settings.confirmTimeout = time.Duration(*account.ConfirmTimeoutSec) * time.Second
	}
	if account.HealthWindowSec != nil && *account.HealthWindowSec >= 60 {
		settings.healthWindow = time.Duration(*account.HealthWindowSec) * time.Second
	}
	if account.HealthRateThreshold != nil && *account.HealthRateThreshold > 0 && *account.HealthRateThreshold <= 1 {
		settings.healthRate = *account.HealthRateThreshold
	}
	if account.HealthMinAttempts != nil && *account.HealthMinAttempts > 0 {
		settings.healthMin = int64(*account.HealthMinAttempts)
	}
	return settings
}

// CheckAttempt is the durable worker-attribution record. Unlike check state,
// it is append-only and therefore remains useful after a worker is deweighted.
type CheckAttempt struct {
	concerns.Model
	JobID              string         `gorm:"uniqueIndex;size:64;not null" json:"job_id"`
	CheckID            int64          `gorm:"index;not null" json:"check_id"`
	MonitorID          int64          `gorm:"index;not null" json:"monitor_id"`
	WorkerNodeID       *int64         `gorm:"index" json:"worker_node_id,omitempty"`
	WorkerNode         *WorkerNode    `json:"worker_node,omitempty"`
	SourceWorkerNodeID *int64         `gorm:"index" json:"source_worker_node_id,omitempty"`
	Kind               string         `gorm:"size:32;not null" json:"kind"`
	State              string         `gorm:"size:32;not null" json:"state"`
	ResultState        string         `gorm:"size:16" json:"result_state"`
	Result             datatypes.JSON `gorm:"type:jsonb;not null;default:'{}'::jsonb" json:"result"`
	StartedAt          *time.Time     `json:"started_at,omitempty"`
	FinishedAt         *time.Time     `json:"finished_at,omitempty"`
	LeaseToken         string         `gorm:"size:64" json:"-"`
	LeaseExpiresAt     *time.Time     `json:"lease_expires_at,omitempty"`
	Deweighted         bool           `gorm:"not null;default:false" json:"deweighted"`
	concerns.Timestamped
}

// DiagnosticAuditEvent is a compact, queryable control-plane audit record.
type DiagnosticAuditEvent struct {
	concerns.Model
	MonitorID    *int64         `gorm:"index" json:"monitor_id,omitempty"`
	WorkerNodeID *int64         `gorm:"index" json:"worker_node_id,omitempty"`
	Kind         string         `gorm:"size:64;index;not null" json:"kind"`
	Metadata     datatypes.JSON `gorm:"type:jsonb;not null;default:'{}'::jsonb" json:"metadata"`
	concerns.Timestamped
}

func auditDiagnostic(tx *gorm.DB, kind string, monitorID, workerID *int64, metadata map[string]interface{}) {
	b, _ := json.Marshal(metadata)
	_ = tx.Create(&DiagnosticAuditEvent{MonitorID: monitorID, WorkerNodeID: workerID, Kind: kind, Metadata: b}).Error
}

// AuditNetworkRecovery is used by the admin incident-response endpoint.
func AuditNetworkRecovery(tx *gorm.DB, workerID int64) {
	auditDiagnostic(tx, "worker.network_problem_force_recover", nil, &workerID, nil)
}

// enqueueDiagnosticDelivery creates the message and its durable delivery task in
// the transition transaction. A transaction advisory lock prevents concurrent
// result frames from leaving duplicate messages when the task dedupe wins.
func enqueueDiagnosticDelivery(tx *gorm.DB, monitor *Monitor, account *Account, tier string, now time.Time) error {
	if account == nil || account.Plan == nil {
		return nil
	}
	if tier == diagnosticHard && !account.Plan.AllowHardAlerts {
		return nil
	}
	var notifications []Notification
	if err := tx.Joins("JOIN notification_groups ON notification_groups.notification_id = notifications.id").
		Where("notifications.account_id = ? AND notifications.enabled AND notification_groups.group_id = ?", account.ID, monitor.GroupID).
		Preload("Contacts", "enabled = ?", true).Find(&notifications).Error; err != nil {
		return err
	}
	for i := range notifications {
		for j := range notifications[i].Contacts {
			contact := notifications[i].Contacts[j]
			method := diagnosticContactMethod(contact.Kind)
			if method == "" || (tier == diagnosticSoft && method != "email" && method != "telegram") {
				continue
			}
			key := fmt.Sprintf("diagnostic:%d:%s:%d:%d", monitor.ID, tier, notifications[i].ID, contact.ID)
			if err := tx.Exec("SELECT pg_advisory_xact_lock(hashtext(?))", key).Error; err != nil {
				return err
			}
			var existing Task
			if err := tx.Where("idempotency_key = ?", key).First(&existing).Error; err == nil {
				continue
			} else if !errors.Is(err, gorm.ErrRecordNotFound) {
				return err
			}
			message := Message{NotificationID: notifications[i].ID, ContactID: contact.ID, Kind: "diagnostic_" + tier, State: TaskStateQueued}
			if err := tx.Create(&message).Error; err != nil {
				return err
			}
			monitorID, messageID := monitor.ID, message.ID
			payload, err := json.Marshal(wire.NotificationTask{AccountID: account.ID, MessageID: messageID, NotificationID: notifications[i].ID, MonitorID: &monitorID, Method: method, Contact: wire.NotificationContact{ID: contact.ID, Kind: contact.Kind, Value: contact.Value, Name: contact.Name}, Subject: diagnosticSubject(monitor, tier), BodyText: diagnosticBody(monitor, tier), BodyMarkdown: diagnosticBody(monitor, tier), BodyHTML: diagnosticBody(monitor, tier), Language: "en", MessageKind: message.Kind})
			if err != nil {
				return err
			}
			if _, err = EnqueueNotificationTaskTx(tx, &EnqueueNotificationTaskInput{AccountID: account.ID, NotificationID: notifications[i].ID, ContactID: contact.ID, MessageID: &messageID, MonitorID: &monitorID, Method: method, Subject: diagnosticSubject(monitor, tier), BodyText: diagnosticBody(monitor, tier), BodyMarkdown: diagnosticBody(monitor, tier), BodyHTML: diagnosticBody(monitor, tier), Language: "en", MessageKind: message.Kind, NotBefore: now, Payload: payload, IdempotencyKey: key}); err != nil {
				return err
			}
		}
	}
	return nil
}

func diagnosticContactMethod(kind string) string {
	switch kind {
	case "email":
		return "email"
	case "telegram_private", "telegram_group":
		return "telegram"
	case "webhook", "mattermost", "sms", "voice":
		return kind
	}
	return ""
}

func diagnosticSubject(monitor *Monitor, tier string) string {
	return fmt.Sprintf("Monitor %s: %s", monitor.Host, tier)
}

func diagnosticBody(monitor *Monitor, tier string) string {
	return fmt.Sprintf("Network diagnostic %s for monitor %s.", tier, monitor.Host)
}

// ConfirmationJobsForWorker atomically leases confirmation jobs assigned to this
// worker or left unassigned by an expired lease. Unassigned attempts still retain
// SourceWorkerNodeID, so the original failing worker can never claim them.
func ConfirmationJobsForWorker(worker *WorkerNode, kinds []string, limit int) ([]wire.CheckJob, error) {
	if worker == nil || worker.AccountID != nil || !worker.SupportsTaskEnvelope() || worker.NetworkProblemActive(time.Now()) || limit < 1 {
		return nil, nil
	}
	var jobs []wire.CheckJob
	err := DB().Transaction(func(tx *gorm.DB) error {
		var attempts []CheckAttempt
		if err := tx.Clauses(SkipLockedClause).Where("(worker_node_id = ? OR worker_node_id IS NULL) AND kind = ? AND state = ?", worker.ID, AttemptKindConfirm, AttemptStateQueued).Order("id").Limit(limit).Find(&attempts).Error; err != nil {
			return err
		}
		for i := range attempts {
			var check Check
			if err := tx.Preload("Monitor").First(&check, attempts[i].CheckID).Error; err != nil {
				continue
			}
			if check.Monitor == nil || !check.Monitor.Enabled {
				continue
			}
			if attempts[i].SourceWorkerNodeID != nil && *attempts[i].SourceWorkerNodeID == worker.ID {
				continue
			}
			if !containsString(kinds, check.Kind) || !containsString(worker.CheckTypes(), check.Kind) {
				continue
			}
			now := time.Now()
			leaseToken := uuid.NewString()
			leaseUntil := now.Add(DefaultTaskLeaseTTL)
			if err := tx.Model(&attempts[i]).Updates(map[string]interface{}{"worker_node_id": worker.ID, "state": AttemptStateLeased, "started_at": now, "lease_token": leaseToken, "lease_expires_at": leaseUntil}).Error; err != nil {
				return err
			}
			if err := tx.Model(&Monitor{}).Where("id = ? AND confirm_state = ?", attempts[i].MonitorID, ConfirmStatePending).Update("confirmed_by_worker_id", worker.ID).Error; err != nil {
				return err
			}
			jobs = append(jobs, wire.CheckJob{JobID: attempts[i].JobID, LeaseToken: leaseToken, CheckID: check.ID, MonitorID: check.MonitorID, Kind: check.Kind, Host: check.Monitor.Host, URL: check.URL, Interval: check.Interval, Settings: json.RawMessage(check.Settings)})
		}
		return nil
	})
	return jobs, err
}

// StartConfirmation creates exactly one targeted confirmation for a new outage.
func StartConfirmation(checkID, sourceWorkerID int64, now time.Time) error {
	return DB().Transaction(func(tx *gorm.DB) error {
		return StartConfirmationTx(tx, checkID, sourceWorkerID, now)
	})
}

// StartConfirmationTx is StartConfirmation's transaction-aware form.
func StartConfirmationTx(tx *gorm.DB, checkID, sourceWorkerID int64, now time.Time) error {
	if tx == nil {
		return errors.New("start confirmation: nil transaction")
	}
	{
		var check Check
		if err := tx.Preload("Monitor.Group.Account.Plan").First(&check, checkID).Error; err != nil {
			return err
		}
		// Confirmations are a paid distributed-check entitlement. Free accounts
		// retain the legacy direct soft alert path and do not consume worker budget.
		if check.Monitor == nil || check.Monitor.Group == nil || check.Monitor.Group.Account == nil || check.Monitor.Group.Account.Plan == nil || !check.Monitor.Group.Account.Plan.Confirmations {
			return nil
		}
		var monitor Monitor
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&monitor, check.MonitorID).Error; err != nil {
			return err
		}
		if monitor.ConfirmState == ConfirmStatePending || monitor.ConfirmState == ConfirmStateConfirmed {
			return nil
		}
		worker, err := confirmationWorkerTx(tx, check.Kind, sourceWorkerID, 0, now)
		if err != nil {
			monitor.ConfirmState, monitor.ConfirmAt = ConfirmStateTimeout, &now
			auditDiagnostic(tx, "check.confirm_unavailable", &monitor.ID, &sourceWorkerID, nil)
			if err := tx.Save(&monitor).Error; err != nil {
				return err
			}
			return enqueueDiagnosticDelivery(tx, &monitor, check.Monitor.Group.Account, diagnosticHard, now)
		}
		monitor.ConfirmState, monitor.ConfirmAt, monitor.ConfirmedByWorkerID = ConfirmStatePending, &now, &worker.ID
		if err := tx.Save(&monitor).Error; err != nil {
			return err
		}
		if err := enqueueDiagnosticDelivery(tx, &monitor, check.Monitor.Group.Account, diagnosticSoft, now); err != nil {
			return err
		}
		attempt := CheckAttempt{JobID: uuid.NewString(), CheckID: check.ID, MonitorID: monitor.ID, WorkerNodeID: &worker.ID, SourceWorkerNodeID: &sourceWorkerID, Kind: AttemptKindConfirm, State: AttemptStateQueued}
		if err := tx.Create(&attempt).Error; err != nil {
			return err
		}
		auditDiagnostic(tx, "check.confirm_assign", &monitor.ID, &worker.ID, map[string]interface{}{"exclude_worker_id": sourceWorkerID, "job_id": attempt.JobID})
		return nil
	}
}

// confirmationWorkerTx selects an independent active platform worker that can
// execute this exact kind. Capability filtering is deliberately performed in
// Go because the JSON capability format also supports legacy rows safely.
func confirmationWorkerTx(tx *gorm.DB, checkKind string, sourceWorkerID, excludeWorkerID int64, now time.Time) (*WorkerNode, error) {
	var workers []WorkerNode
	if err := tx.Where("id <> ? AND id <> ? AND account_id IS NULL AND status = 'active' AND (network_problems = FALSE OR network_problems_until <= ? OR network_problems_until IS NULL)", sourceWorkerID, excludeWorkerID, now).Order("id").Find(&workers).Error; err != nil {
		return nil, err
	}
	for i := range workers {
		if workers[i].SupportsTaskEnvelope() && containsString(workers[i].CheckTypes(), checkKind) {
			return &workers[i], nil
		}
	}
	return nil, gorm.ErrRecordNotFound
}

// ApplyDiagnosticResult resolves a targeted attempt once. Duplicate reports are ignored.
func ApplyDiagnosticResult(report wire.CheckResultReport, worker *WorkerNode, now time.Time) (bool, error) {
	if report.JobID == "" || worker == nil {
		return false, nil
	}
	handled := true
	err := DB().Transaction(func(tx *gorm.DB) error {
		return ApplyDiagnosticResultTx(tx, report, worker, now, &handled)
	})
	return handled, err
}

// ApplyDiagnosticResultTx resolves a diagnostic attempt within the caller's
// transaction. handled distinguishes a normal check result from a diagnostic.
func ApplyDiagnosticResultTx(tx *gorm.DB, report wire.CheckResultReport, worker *WorkerNode, now time.Time, handled *bool) error {
	if tx == nil {
		return errors.New("apply diagnostic: nil transaction")
	}
	if handled == nil {
		return errors.New("apply diagnostic: nil handled result")
	}
	*handled = true
	{
		var attempt CheckAttempt
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("job_id = ?", report.JobID).First(&attempt).Error; err != nil {
			if err == gorm.ErrRecordNotFound {
				*handled = false
				return nil
			}
			return err
		}
		if attempt.Kind != AttemptKindConfirm {
			*handled = false
			return nil
		}
		if attempt.State == AttemptStateFinished {
			*handled = false
			return nil
		}
		if attempt.State != AttemptStateLeased || attempt.WorkerNodeID == nil || *attempt.WorkerNodeID != worker.ID {
			return gorm.ErrRecordNotFound
		}
		if attempt.LeaseToken == "" || report.LeaseToken == "" || report.LeaseToken != attempt.LeaseToken || attempt.LeaseExpiresAt == nil || !attempt.LeaseExpiresAt.After(now) {
			return errors.New("apply diagnostic: lease token is invalid or expired")
		}
		payload, _ := json.Marshal(report)
		deweighted := worker.NetworkProblemActive(now)
		if err := tx.Model(&attempt).Where("state = ? AND lease_token = ? AND lease_expires_at > ?", AttemptStateLeased, report.LeaseToken, now).Updates(map[string]interface{}{"state": AttemptStateFinished, "result_state": report.State, "result": payload, "finished_at": now, "lease_token": "", "lease_expires_at": nil, "deweighted": deweighted}).Error; err != nil {
			return err
		}
		var monitor Monitor
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&monitor, attempt.MonitorID).Error; err != nil {
			return err
		}
		if attempt.Kind == AttemptKindConfirm && monitor.ConfirmState == ConfirmStatePending {
			if report.State == stateERR || report.State == stateFail {
				monitor.ConfirmState = ConfirmStateConfirmed
				if err := enqueueDiagnosticDelivery(tx, &monitor, monitorAccount(tx, &monitor), diagnosticHard, now); err != nil {
					return err
				}
			} else {
				monitor.ConfirmState = ConfirmStateNone
				if err := enqueueDiagnosticDelivery(tx, &monitor, monitorAccount(tx, &monitor), diagnosticRecovery, now); err != nil {
					return err
				}
			}
			if err := tx.Save(&monitor).Error; err != nil {
				return err
			}
			auditDiagnostic(tx, "check.confirm_result", &monitor.ID, &worker.ID, map[string]interface{}{"state": report.State, "deweighted": deweighted})
		}
		return nil
	}
}

func monitorAccount(tx *gorm.DB, monitor *Monitor) *Account {
	var group Group
	if err := tx.Preload("Account.Plan").First(&group, monitor.GroupID).Error; err != nil {
		return nil
	}
	return group.Account
}

// RecoverDiagnostic clears a completed hard escalation only once and queues the
// corresponding recovery tasks in the same transaction.
func RecoverDiagnostic(checkID int64, now time.Time) error {
	return DB().Transaction(func(tx *gorm.DB) error {
		return RecoverDiagnosticTx(tx, checkID, now)
	})
}

// RecoverDiagnosticTx is RecoverDiagnostic's transaction-aware form.
func RecoverDiagnosticTx(tx *gorm.DB, checkID int64, now time.Time) error {
	if tx == nil {
		return errors.New("recover diagnostic: nil transaction")
	}
	{
		var check Check
		if err := tx.Preload("Monitor.Group.Account.Plan").First(&check, checkID).Error; err != nil {
			return err
		}
		if check.Monitor == nil || check.Monitor.Group == nil || check.Monitor.Group.Account == nil {
			return nil
		}
		var monitor Monitor
		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&monitor, check.MonitorID).Error; err != nil {
			return err
		}
		if monitor.ConfirmState != ConfirmStateConfirmed && monitor.ConfirmState != ConfirmStateTimeout {
			return nil
		}
		monitor.ConfirmState = ConfirmStateNone
		if err := tx.Save(&monitor).Error; err != nil {
			return err
		}
		auditDiagnostic(tx, "check.confirm_recovery", &monitor.ID, nil, nil)
		return enqueueDiagnosticDelivery(tx, &monitor, check.Monitor.Group.Account, diagnosticRecovery, now)
	}
}

// NetworkDiagnosticsTick expires confirmations and derives worker health from durable attempts.
func NetworkDiagnosticsTick(now time.Time) error {
	return DB().Transaction(func(tx *gorm.DB) error {
		if err := reapExpiredConfirmationAttemptsTx(tx, now); err != nil {
			return err
		}
		var monitors []Monitor
		if err := tx.Clauses(SkipLockedClause).Preload("Group.Account.Plan").Where("confirm_state = ?", ConfirmStatePending).Find(&monitors).Error; err != nil {
			return err
		}
		for i := range monitors {
			settings := settingsForAccount(monitors[i].Group.Account)
			if monitors[i].ConfirmAt == nil || monitors[i].ConfirmAt.After(now.Add(-settings.confirmTimeout)) {
				continue
			}
			if err := tx.Model(&monitors[i]).Update("confirm_state", ConfirmStateTimeout).Error; err != nil {
				return err
			}
			auditDiagnostic(tx, "check.confirm_timeout", &monitors[i].ID, nil, nil)
			if err := enqueueDiagnosticDelivery(tx, &monitors[i], monitors[i].Group.Account, diagnosticHard, now); err != nil {
				return err
			}
		}
		var workers []WorkerNode
		if err := tx.Find(&workers).Error; err != nil {
			return err
		}
		for i := range workers {
			var total, failed int64
			// Operated workers serve accounts on different plans. Use the most
			// sensitive entitled setting among their recent attempts.
			settings := diagnosticSettings{healthWindow: 5 * time.Minute, healthRate: .5, healthMin: 10}
			q := tx.Model(&CheckAttempt{}).Where("worker_node_id = ? AND finished_at >= ?", workers[i].ID, now.Add(-settings.healthWindow))
			q.Count(&total)
			q.Where("result_state IN ?", []string{stateERR, stateFail}).Count(&failed)
			flagged := total >= settings.healthMin && float64(failed)/float64(total) >= settings.healthRate
			updates := map[string]interface{}{"last_total_count": total, "last_failure_count": failed}
			if flagged {
				updates["network_problems"] = true
				updates["network_problems_until"] = now.Add(10 * time.Minute)
			}
			if workers[i].NetworkProblems && workers[i].NetworkProblemsUntil != nil && workers[i].NetworkProblemsUntil.Before(now) && !flagged {
				updates["network_problems"] = false
				updates["network_problems_until"] = nil
				auditDiagnostic(tx, "worker.network_problem_unflag", nil, &workers[i].ID, nil)
			}
			if flagged && !workers[i].NetworkProblems {
				auditDiagnostic(tx, "worker.network_problem_flag", nil, &workers[i].ID, map[string]interface{}{"failures": failed, "total": total})
			}
			if err := tx.Model(&workers[i]).Updates(updates).Error; err != nil {
				return err
			}
			_ = influx.WriteOne("worker_health", map[string]string{"worker_id": workers[i].WorkerID}, map[string]interface{}{"failures": failed, "total": total, "failure_rate": float64(failed) / float64(maxInt64(total, 1))})
		}
		return nil
	})
}

func reapExpiredConfirmationAttemptsTx(tx *gorm.DB, now time.Time) error {
	var attempts []CheckAttempt
	if err := tx.Clauses(SkipLockedClause).Where("kind = ? AND state = ? AND lease_expires_at <= ?", AttemptKindConfirm, AttemptStateLeased, now).Find(&attempts).Error; err != nil {
		return err
	}
	for i := range attempts {
		var check Check
		if err := tx.First(&check, attempts[i].CheckID).Error; err != nil {
			return err
		}
		oldWorkerID := int64(0)
		if attempts[i].WorkerNodeID != nil {
			oldWorkerID = *attempts[i].WorkerNodeID
		}
		sourceWorkerID := int64(0)
		if attempts[i].SourceWorkerNodeID != nil {
			sourceWorkerID = *attempts[i].SourceWorkerNodeID
		}
		worker, err := confirmationWorkerTx(tx, check.Kind, sourceWorkerID, oldWorkerID, now)
		if errors.Is(err, gorm.ErrRecordNotFound) {
			// No replacement is available now. Remove the stale assignment so a
			// later capable independent worker can claim this queued attempt.
			if err := tx.Model(&attempts[i]).Updates(map[string]interface{}{"worker_node_id": nil, "state": AttemptStateQueued, "lease_token": "", "lease_expires_at": nil}).Error; err != nil {
				return err
			}
			if err := tx.Model(&Monitor{}).Where("id = ? AND confirm_state = ?", attempts[i].MonitorID, ConfirmStatePending).Update("confirmed_by_worker_id", nil).Error; err != nil {
				return err
			}
			continue
		}
		if err != nil {
			return err
		}
		if err := tx.Model(&attempts[i]).Updates(map[string]interface{}{"worker_node_id": worker.ID, "state": AttemptStateQueued, "lease_token": "", "lease_expires_at": nil}).Error; err != nil {
			return err
		}
		if err := tx.Model(&Monitor{}).Where("id = ? AND confirm_state = ?", attempts[i].MonitorID, ConfirmStatePending).Update("confirmed_by_worker_id", worker.ID).Error; err != nil {
			return err
		}
	}
	return nil
}

func maxInt64(a, b int64) int64 {
	if a > b {
		return a
	}
	return b
}