oci-portal/internal/service/task.go

package service

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

	"github.com/robfig/cron/v3"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"oci-portal/internal/model"
	"oci-portal/internal/oci"
)

// taskRunTimeout 是单次任务执行的超时时间。
const taskRunTimeout = 10 * time.Minute

// taskLogKeep 是每个任务保留的执行日志条数。
const taskLogKeep = 100

// defaultSnatchIPWait 是抢机成功后等待公网 IPv4 分配的总预算(全部实例共享,
// 实例从创建到 VNIC 就绪通常 1~2 分钟);taskRunTimeout 余量内,超时降级不阻塞通知。
const defaultSnatchIPWait = 3 * time.Minute

// snatchIPPollInterval 是等待公网 IP 的轮询间隔。
const snatchIPPollInterval = 10 * time.Second

// TaskService 管理后台任务的存储、cron 调度与执行。
type TaskService struct {
	db       *gorm.DB
	configs  *OciConfigService
	notifier *Notifier
	settings *SettingService
	cron     *cron.Cron
	// aiGateway 供 AI 探测任务执行渠道探测,由 main 装配(可为 nil)
	aiGateway *AiGatewayService
	// snatchIPWait 是抢机成功后等待公网 IP 的预算,测试置 0 跳过等待
	snatchIPWait time.Duration

	mu      sync.Mutex
	entries map[uint]cron.EntryID
	// runMu 让租户删除等待在途执行结束,并阻止新执行读取删除前的任务快照。
	runMu sync.RWMutex
	// runningMu 保护 running:同一任务不允许并发重复执行
	// (cron 重叠触发静默跳过,手动触发返回 ErrTaskRunning)
	runningMu sync.Mutex
	running   map[uint]bool
	// runWG 追踪手动触发的后台执行,Stop 时等待收尾
	runWG sync.WaitGroup

	// snatchVarsMu 保护 snatchVars:runSnatch 抢满时写入成功通知变量,
	// execute 组装执行后快照时取走(一次性)
	snatchVarsMu sync.Mutex
	snatchVars   map[uint]map[string]string
}

// NewTaskService 组装依赖；notifier 传 nil 表示整体关闭通知，
// settings 供发送前按事件类型过滤（nil 视为全开）。调用 Start 后开始调度。
func NewTaskService(db *gorm.DB, configs *OciConfigService, notifier *Notifier, settings *SettingService) *TaskService {
	return &TaskService{
		db:           db,
		configs:      configs,
		notifier:     notifier,
		settings:     settings,
		cron:         cron.New(),
		entries:      map[uint]cron.EntryID{},
		running:      map[uint]bool{},
		snatchIPWait: defaultSnatchIPWait,
		snatchVars:   map[uint]map[string]string{},
	}
}

// AttachAiGateway 注入 AI 网关服务,启用 AI 探测任务的执行与自动同步。
func (s *TaskService) AttachAiGateway(gw *AiGatewayService) { s.aiGateway = gw }

// Start 加载全部 active 任务注册调度并启动 cron。
func (s *TaskService) Start() error {
	var tasks []model.Task
	if err := s.db.Where("status = ?", model.TaskStatusActive).Find(&tasks).Error; err != nil {
		return fmt.Errorf("load active tasks: %w", err)
	}
	for i := range tasks {
		if err := s.schedule(&tasks[i]); err != nil {
			return err
		}
	}
	s.cron.Start()
	return nil
}

// Stop 停止调度并等待执行中的任务收尾(含手动触发的后台执行),
// 保证任务产生的异步通知都已进入 Notifier 的等待队列。
func (s *TaskService) Stop() {
	<-s.cron.Stop().Done()
	s.runWG.Wait()
}

// healthCheckPayload 是测活任务参数；ociConfigIds 为空表示全部配置。
type healthCheckPayload struct {
	OciConfigIDs []uint `json:"ociConfigIds"`
}

// costPayload 是成本同步任务参数；ociConfigIds 为空表示全部配置，
// 免费类别的配置在执行时跳过，不发起 Usage API 请求。
type costPayload struct {
	OciConfigIDs []uint `json:"ociConfigIds"`
}

// snatchPayload 是抢机任务参数；count 为剩余台数（创建时即目标台数，
// 每次执行把剩余写回，直到抢满），totalCount 固定为创建时的目标台数，
// 供前端计算进度（旧任务缺省时前端回退用 count）。
// authFailCount 为连续 NotAuthenticated 失败计数，达阈值任务熔断停止。
type snatchPayload struct {
	OciConfigID   uint                    `json:"ociConfigId"`
	Count         int                     `json:"count"`
	TotalCount    int                     `json:"totalCount,omitempty"`
	AuthFailCount int                     `json:"authFailCount,omitempty"`
	Instance      oci.CreateInstanceInput `json:"instance"`
}

// CreateTaskInput 是创建任务的输入。
type CreateTaskInput struct {
	Name     string
	Type     string
	CronExpr string
	Payload  json.RawMessage
}

// CreateTask 校验并保存任务,立即进入调度;AI 探测任务全局唯一(系统自动管理)。
func (s *TaskService) CreateTask(ctx context.Context, in CreateTaskInput) (*model.Task, error) {
	if in.Name == "" {
		return nil, fmt.Errorf("create task: name is required")
	}
	if _, err := cron.ParseStandard(in.CronExpr); err != nil {
		return nil, fmt.Errorf("create task: invalid cron %q: %w", in.CronExpr, err)
	}
	if err := validateTaskPayload(in.Type, in.Payload); err != nil {
		return nil, err
	}
	if err := s.ensureAiProbeUnique(ctx, in.Type); err != nil {
		return nil, err
	}
	task := &model.Task{
		Name:     in.Name,
		Type:     in.Type,
		CronExpr: in.CronExpr,
		Payload:  string(normalizeSnatchPayload(in.Type, in.Payload)),
		Status:   model.TaskStatusActive,
	}
	if err := s.db.WithContext(ctx).Create(task).Error; err != nil {
		return nil, fmt.Errorf("create task: %w", err)
	}
	if err := s.schedule(task); err != nil {
		return nil, err
	}
	return task, nil
}

// ensureAiProbeUnique 拒绝重复创建 AI 探测任务(该类型随渠道数量自动管理)。
func (s *TaskService) ensureAiProbeUnique(ctx context.Context, taskType string) error {
	if taskType != model.TaskTypeAiProbe {
		return nil
	}
	var n int64
	if err := s.db.WithContext(ctx).Model(&model.Task{}).
		Where("type = ?", model.TaskTypeAiProbe).Count(&n).Error; err != nil {
		return err
	}
	if n > 0 {
		return fmt.Errorf("create task: AI 探测任务已存在,由系统自动管理")
	}
	return nil
}

// normalizeSnatchPayload 给抢机 payload 补全目标台数：count 默认 1、
// totalCount 缺省时固定为创建时的 count，供进度展示；解析失败原样返回
// （validateTaskPayload 已在前面拦截非法 JSON）。
func normalizeSnatchPayload(taskType string, payload json.RawMessage) json.RawMessage {
	if taskType != model.TaskTypeSnatch {
		return payload
	}
	var p snatchPayload
	if err := json.Unmarshal(payload, &p); err != nil {
		return payload
	}
	if p.Count <= 0 {
		p.Count = 1
	}
	if p.TotalCount <= 0 {
		p.TotalCount = p.Count
	}
	out, err := json.Marshal(p)
	if err != nil {
		return payload
	}
	return out
}

// mergeSnatchPayload 把编辑提交的 count 视为**新目标台数**:按旧 payload 的
// 已完成数换算剩余、保留连续鉴权失败计数,目标不大于已完成数时拒绝。
// 非抢机任务原样返回;旧 payload 不可解析时按新建任务归一化。
func mergeSnatchPayload(task *model.Task, incoming json.RawMessage) (json.RawMessage, error) {
	if task.Type != model.TaskTypeSnatch {
		return incoming, nil
	}
	var old, next snatchPayload
	if err := json.Unmarshal([]byte(task.Payload), &old); err != nil {
		return normalizeSnatchPayload(task.Type, incoming), nil
	}
	if err := json.Unmarshal(incoming, &next); err != nil {
		return nil, fmt.Errorf("update task: invalid payload: %w", err)
	}
	oldTotal := old.TotalCount
	if oldTotal <= 0 {
		oldTotal = old.Count
	}
	done := oldTotal - old.Count
	if next.Count <= done {
		return nil, fmt.Errorf("%w(已完成 %d 台)", ErrSnatchTargetTooLow, done)
	}
	next.TotalCount = next.Count
	next.Count -= done
	next.AuthFailCount = old.AuthFailCount
	out, err := json.Marshal(next)
	if err != nil {
		return nil, fmt.Errorf("update task: marshal payload: %w", err)
	}
	return out, nil
}

// validateTaskPayload 按任务类型校验参数 JSON。
func validateTaskPayload(taskType string, payload json.RawMessage) error {
	switch taskType {
	case model.TaskTypeHealthCheck:
		var p healthCheckPayload
		if len(payload) > 0 {
			if err := json.Unmarshal(payload, &p); err != nil {
				return fmt.Errorf("create task: invalid payload: %w", err)
			}
		}
		return nil
	case model.TaskTypeCost:
		var p costPayload
		if len(payload) > 0 {
			if err := json.Unmarshal(payload, &p); err != nil {
				return fmt.Errorf("create task: invalid payload: %w", err)
			}
		}
		return nil
	case model.TaskTypeSnatch:
		var p snatchPayload
		if err := json.Unmarshal(payload, &p); err != nil {
			return fmt.Errorf("create task: invalid payload: %w", err)
		}
		if p.OciConfigID == 0 {
			return fmt.Errorf("create task: snatch payload requires ociConfigId")
		}
		return validateCreateInstance(p.Instance)
	case model.TaskTypeAiProbe:
		return nil // 无参数:探测全部渠道
	default:
		return fmt.Errorf("create task: unsupported type %q", taskType)
	}
}

// UpdateTaskInput 是更新任务的输入；nil 字段不修改。
type UpdateTaskInput struct {
	Name     *string
	CronExpr *string
	Payload  json.RawMessage
	Status   *string
}

// UpdateTask 修改任务并重新调度;写入用 updated_at 条件更新,
// 防止陈旧快照整行覆盖并发执行刚落库的状态与进度。
func (s *TaskService) UpdateTask(ctx context.Context, id uint, in UpdateTaskInput) (*model.Task, error) {
	task, err := s.GetTask(ctx, id)
	if err != nil {
		return nil, err
	}
	if err := applyTaskUpdate(task, in); err != nil {
		return nil, err
	}
	fresh, err := s.persistTaskUpdate(ctx, task)
	if err != nil {
		return nil, err
	}
	s.unschedule(fresh.ID)
	if fresh.Status == model.TaskStatusActive {
		if err := s.schedule(fresh); err != nil {
			return nil, err
		}
	}
	return fresh, nil
}

// persistTaskUpdate 只写用户可编辑列;零命中说明执行侧已并发落库,返回冲突。
func (s *TaskService) persistTaskUpdate(ctx context.Context, task *model.Task) (*model.Task, error) {
	res := s.db.WithContext(ctx).Model(&model.Task{}).
		Where("id = ? AND updated_at = ?", task.ID, task.UpdatedAt).
		Updates(map[string]any{
			"name": task.Name, "cron_expr": task.CronExpr,
			"payload": task.Payload, "status": task.Status,
		})
	if res.Error != nil {
		return nil, fmt.Errorf("update task %d: %w", task.ID, res.Error)
	}
	if res.RowsAffected == 0 {
		return nil, ErrTaskConflict
	}
	// 重读用新变量:gorm 扫描 NULL 列到已有值的结构体时会保留旧值
	var fresh model.Task
	if err := s.db.WithContext(ctx).First(&fresh, task.ID).Error; err != nil {
		return nil, fmt.Errorf("reload task %d: %w", task.ID, err)
	}
	return &fresh, nil
}

func applyTaskUpdate(task *model.Task, in UpdateTaskInput) error {
	if in.Name != nil {
		task.Name = *in.Name
	}
	if in.CronExpr != nil {
		if _, err := cron.ParseStandard(*in.CronExpr); err != nil {
			return fmt.Errorf("update task: invalid cron %q: %w", *in.CronExpr, err)
		}
		task.CronExpr = *in.CronExpr
	}
	if len(in.Payload) > 0 {
		if err := validateTaskPayload(task.Type, in.Payload); err != nil {
			return err
		}
		merged, err := mergeSnatchPayload(task, in.Payload)
		if err != nil {
			return err
		}
		task.Payload = string(merged)
	}
	if in.Status != nil {
		if *in.Status != model.TaskStatusActive && *in.Status != model.TaskStatusPaused {
			return fmt.Errorf("update task: status must be active or paused")
		}
		if task.Status == model.TaskStatusFailed && *in.Status == model.TaskStatusActive {
			resetSnatchAuthFail(task)
		}
		task.Status = *in.Status
	}
	return nil
}

// resetSnatchAuthFail 清零抢机 payload 的连续鉴权失败计数，
// 供 failed 任务重新启用时调用，避免一恢复调度就再次熔断；非抢机任务不处理。
func resetSnatchAuthFail(task *model.Task) {
	if task.Type != model.TaskTypeSnatch {
		return
	}
	var p snatchPayload
	if err := json.Unmarshal([]byte(task.Payload), &p); err != nil {
		return
	}
	p.AuthFailCount = 0
	writeSnatchPayload(task, &p)
}

// ListTasks 返回全部任务。
func (s *TaskService) ListTasks(ctx context.Context) ([]model.Task, error) {
	tasks := make([]model.Task, 0)
	if err := s.db.WithContext(ctx).Order("id").Find(&tasks).Error; err != nil {
		return nil, fmt.Errorf("list tasks: %w", err)
	}
	return tasks, nil
}

// GetTask 返回单个任务。
func (s *TaskService) GetTask(ctx context.Context, id uint) (*model.Task, error) {
	var task model.Task
	if err := s.db.WithContext(ctx).First(&task, id).Error; err != nil {
		return nil, fmt.Errorf("find task %d: %w", id, err)
	}
	return &task, nil
}

// DeleteTask 注销调度并删除任务与其日志;AI 探测任务由系统按渠道数量
// 自动创建/删除,拒绝手动删除。
func (s *TaskService) DeleteTask(ctx context.Context, id uint) error {
	task, err := s.GetTask(ctx, id)
	if err != nil {
		return err
	}
	if task.Type == model.TaskTypeAiProbe {
		return fmt.Errorf("delete task: AI 探测任务由系统自动管理,删除最后一个渠道时自动移除")
	}
	return s.removeTask(ctx, id)
}

// removeTask 注销调度并删除任务与其日志(内部路径,不做类型限制)。
func (s *TaskService) removeTask(ctx context.Context, id uint) error {
	s.unschedule(id)
	if err := s.db.WithContext(ctx).Delete(&model.Task{}, id).Error; err != nil {
		return fmt.Errorf("delete task %d: %w", id, err)
	}
	if err := s.db.WithContext(ctx).Where("task_id = ?", id).Delete(&model.TaskLog{}).Error; err != nil {
		return fmt.Errorf("delete task %d logs: %w", id, err)
	}
	return nil
}

// TaskLogs 返回任务最近的执行日志（时间倒序）。
func (s *TaskService) TaskLogs(ctx context.Context, id uint, limit int) ([]model.TaskLog, error) {
	if limit <= 0 || limit > taskLogKeep {
		limit = 50
	}
	logs := make([]model.TaskLog, 0)
	err := s.db.WithContext(ctx).Where("task_id = ?", id).
		Order("id desc").Limit(limit).Find(&logs).Error
	if err != nil {
		return nil, fmt.Errorf("list task %d logs: %w", id, err)
	}
	return logs, nil
}

// RunTaskNow 同步执行一次任务并返回本次日志(内部与测试使用;API 走 TriggerTask)。
func (s *TaskService) RunTaskNow(ctx context.Context, id uint) (*model.TaskLog, error) {
	if _, err := s.GetTask(ctx, id); err != nil {
		return nil, err
	}
	return s.execute(id), nil
}

// ErrTaskRunning 表示任务已有一次执行在途,拒绝重复触发。
var ErrTaskRunning = errors.New("任务正在执行中,请稍候")

// ErrTaskConflict 表示任务在编辑期间被并发修改(如执行结果落库),须刷新重试。
var ErrTaskConflict = errors.New("任务状态已变化,请刷新后重试")

// ErrSnatchTargetTooLow 表示编辑抢机任务时新目标台数不大于已完成数量。
var ErrSnatchTargetTooLow = errors.New("目标台数须大于已完成数量")

// TriggerTask 异步触发一次任务执行并立即返回;执行结果照常落任务日志与通知,
// 由前端轮询呈现。同一任务在途时返回 ErrTaskRunning。
func (s *TaskService) TriggerTask(ctx context.Context, id uint) error {
	if _, err := s.GetTask(ctx, id); err != nil {
		return err
	}
	if !s.beginRun(id) {
		return ErrTaskRunning
	}
	s.runWG.Add(1)
	go func() {
		defer s.runWG.Done()
		s.runHeld(id)
	}()
	return nil
}

// beginRun 抢占任务的执行权;在途时返回 false。
func (s *TaskService) beginRun(id uint) bool {
	s.runningMu.Lock()
	defer s.runningMu.Unlock()
	if s.running[id] {
		return false
	}
	s.running[id] = true
	return true
}

func (s *TaskService) endRun(id uint) {
	s.runningMu.Lock()
	delete(s.running, id)
	s.runningMu.Unlock()
}

// schedule 把任务注册进 cron 调度。
func (s *TaskService) schedule(task *model.Task) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if _, ok := s.entries[task.ID]; ok {
		return nil
	}
	taskID := task.ID
	entry, err := s.cron.AddFunc(task.CronExpr, func() { s.execute(taskID) })
	if err != nil {
		return fmt.Errorf("schedule task %d: %w", task.ID, err)
	}
	s.entries[task.ID] = entry
	return nil
}

// unschedule 把任务移出 cron 调度。
func (s *TaskService) unschedule(taskID uint) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if entry, ok := s.entries[taskID]; ok {
		s.cron.Remove(entry)
		delete(s.entries, taskID)
	}
}

// ApplyTenantCleanup 在租户删除事务提交后注销已删除任务并对齐 AI 探测任务。
func (s *TaskService) ApplyTenantCleanup(ctx context.Context, taskIDs []uint, channelsGone bool) {
	for _, id := range taskIDs {
		s.unschedule(id)
		s.forgetSnatchVars(id)
	}
	if channelsGone {
		s.SyncAiProbeTask(ctx)
	}
}

func (s *TaskService) forgetSnatchVars(taskID uint) {
	s.snatchVarsMu.Lock()
	delete(s.snatchVars, taskID)
	s.snatchVarsMu.Unlock()
}

func (s *TaskService) lockTenantCleanup() func() {
	s.runMu.Lock()
	return s.runMu.Unlock
}

// execute 执行一次任务:同一任务已有执行在途时静默跳过(cron 重叠触发防抖)。
func (s *TaskService) execute(taskID uint) *model.TaskLog {
	if !s.beginRun(taskID) {
		return nil
	}
	return s.runHeld(taskID)
}

// runHeld 在已持有执行权的前提下完成一次执行:加载 → 分派 → 更新任务状态并写日志,
// 前后状态交给通知判定,只在状态变化时推送。
func (s *TaskService) runHeld(taskID uint) *model.TaskLog {
	defer s.endRun(taskID)
	s.runMu.RLock()
	defer s.runMu.RUnlock()
	return s.executeLocked(taskID)
}

func (s *TaskService) executeLocked(taskID uint) *model.TaskLog {
	ctx, cancel := context.WithTimeout(context.Background(), taskRunTimeout)
	defer cancel()
	task, err := s.GetTask(ctx, taskID)
	if err != nil {
		return nil
	}
	prev := taskSnapshot{Name: task.Name, Status: task.Status, LastError: task.LastError}
	start := time.Now()
	message, runErr := s.run(ctx, task)
	now := time.Now()
	task.LastRunAt = &now
	task.RunCount++
	task.LastError = ""
	if runErr != nil {
		task.LastError = oci.CompactError(runErr)
		message = task.LastError
	}
	if !s.storeTaskRun(ctx, task) {
		return nil
	}
	cur := taskSnapshot{Name: task.Name, Status: task.Status, LastError: task.LastError,
		Message: message, SnatchVars: s.takeSnatchVars(task.ID)}
	s.notify(notifyEvents(prev, cur))
	return s.appendLog(task.ID, runErr == nil, message, time.Since(start))
}

func (s *TaskService) storeTaskRun(ctx context.Context, task *model.Task) bool {
	stored, err := s.persistTaskRun(ctx, task)
	if err != nil {
		log.Printf("persist task run: %v", err)
		return false
	}
	return stored
}

func (s *TaskService) persistTaskRun(ctx context.Context, task *model.Task) (bool, error) {
	updates := map[string]any{
		"status": task.Status, "last_run_at": task.LastRunAt,
		"last_error": task.LastError, "run_count": task.RunCount,
	}
	if task.Type == model.TaskTypeSnatch {
		updates["payload"] = task.Payload
	}
	res := s.db.WithContext(ctx).Model(&model.Task{}).
		Where("id = ? AND updated_at = ?", task.ID, task.UpdatedAt).Updates(updates)
	if res.Error != nil {
		return false, fmt.Errorf("persist task %d run: %w", task.ID, res.Error)
	}
	return res.RowsAffected == 1, nil
}

// run 按类型分派任务执行。
func (s *TaskService) run(ctx context.Context, task *model.Task) (string, error) {
	switch task.Type {
	case model.TaskTypeHealthCheck:
		return s.runHealthCheck(ctx, task)
	case model.TaskTypeCost:
		return s.runCost(ctx, task)
	case model.TaskTypeSnatch:
		return s.runSnatch(ctx, task)
	case model.TaskTypeAiProbe:
		return s.runAiProbe(ctx)
	default:
		return "", fmt.Errorf("unsupported task type %q", task.Type)
	}
}

// runAiProbe 逐渠道探测 AI 网关号池;网关未装配时报错。
func (s *TaskService) runAiProbe(ctx context.Context) (string, error) {
	if s.aiGateway == nil {
		return "", fmt.Errorf("ai gateway not attached")
	}
	msg, err := s.aiGateway.ProbeAll(ctx)
	if err == nil {
		s.warnDeprecatingModels(ctx)
	}
	return msg, err
}

// warnDeprecatingModels 对 30 天内即将退役或弃用的在池模型发 Telegram 提醒;
// 随每日探测执行,模型退役被同步剔除后自动停止,受通知管理 model_deprecated 开关控制。
func (s *TaskService) warnDeprecatingModels(ctx context.Context) {
	if s.notifier == nil {
		return
	}
	if s.settings != nil && !s.settings.NotifyEventEnabled(ctx, "model_deprecated") {
		return
	}
	names, err := s.aiGateway.DeprecatingModels(ctx, 30*24*time.Hour)
	if err != nil || len(names) == 0 {
		return
	}
	s.notifier.SendTemplateAsync("model_deprecated", map[string]string{"models": strings.Join(names, "\n")})
}

// runHealthCheck 对范围内的配置逐个测活，汇总结果并触发失联通知。
func (s *TaskService) runHealthCheck(ctx context.Context, task *model.Task) (string, error) {
	var p healthCheckPayload
	if task.Payload != "" {
		if err := json.Unmarshal([]byte(task.Payload), &p); err != nil {
			return "", fmt.Errorf("parse payload: %w", err)
		}
	}
	ids, err := s.targetConfigIDs(ctx, p.OciConfigIDs)
	if err != nil {
		return "", err
	}
	alive := 0
	var deadAliases, failures []string
	for _, id := range ids {
		cfg, _, err := s.configs.Verify(ctx, id)
		ok := err == nil && cfg.AliveStatus == model.AliveStatusAlive
		s.saveCheckSnapshot(ctx, id, ok)
		if !ok {
			deadAliases = append(deadAliases, configAlias(cfg, id))
			failures = append(failures, fmt.Sprintf("#%d %s", id, verifyFailReason(cfg, err)))
			continue
		}
		alive++
	}
	s.notifyDeadAliases(ctx, task.ID, deadAliases)
	msg := fmt.Sprintf("checked %d: %d alive, %d dead", len(ids), alive, len(deadAliases))
	if len(failures) > 0 {
		msg += "; " + strings.Join(failures, "; ")
	}
	return msg, nil
}

// configAlias 返回配置别名，配置加载失败时退回 #ID 表示。
func configAlias(cfg *model.OciConfig, id uint) string {
	if cfg != nil && cfg.Alias != "" {
		return cfg.Alias
	}
	return fmt.Sprintf("#%d", id)
}

func verifyFailReason(cfg *model.OciConfig, err error) string {
	if err != nil {
		return oci.CompactError(err)
	}
	return cfg.LastError
}

// saveCheckSnapshot 覆盖写入测活快照；仅存活时刷新实例数（默认区域口径），
// 失联时保留上次实例数，避免总览 KPI 因 key 失效而抖动。
func (s *TaskService) saveCheckSnapshot(ctx context.Context, cfgID uint, alive bool) {
	status := model.AliveStatusDead
	if alive {
		status = model.AliveStatusAlive
	}
	snap := model.CheckSnapshot{OciConfigID: cfgID, AliveStatus: status, CheckedAt: time.Now()}
	cols := []string{"alive_status", "checked_at"}
	if alive {
		if instances, err := s.configs.Instances(ctx, cfgID, "", ""); err == nil {
			snap.InstanceCount = len(instances)
			cols = append(cols, "instance_count")
		}
	}
	s.db.WithContext(ctx).Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "oci_config_id"}},
		DoUpdates: clause.AssignmentColumns(cols),
	}).Create(&snap)
}

// runCost 对范围内配置同步近 7 天每日成本快照，免费类别跳过。
func (s *TaskService) runCost(ctx context.Context, task *model.Task) (string, error) {
	var p costPayload
	if task.Payload != "" {
		if err := json.Unmarshal([]byte(task.Payload), &p); err != nil {
			return "", fmt.Errorf("parse payload: %w", err)
		}
	}
	ids, err := s.targetConfigIDs(ctx, p.OciConfigIDs)
	if err != nil {
		return "", err
	}
	synced, skipped := 0, 0
	var failures []string
	for _, id := range ids {
		switch err := s.syncCostSnapshot(ctx, id); {
		case err == errFreeAccountSkipped:
			skipped++
		case err != nil:
			failures = append(failures, fmt.Sprintf("#%d %v", id, err))
		default:
			synced++
		}
	}
	msg := fmt.Sprintf("synced usage for %d tenants, skipped %d free", synced, skipped)
	if len(failures) > 0 {
		msg += "; " + strings.Join(failures, "; ")
	}
	return msg, nil
}

// errFreeAccountSkipped 标记成本同步因免费类别被跳过。
var errFreeAccountSkipped = fmt.Errorf("free account skipped")

// syncCostSnapshot 拉取单配置近 7 天每日成本并按天覆盖写入快照。
func (s *TaskService) syncCostSnapshot(ctx context.Context, cfgID uint) error {
	cfg, err := s.configs.Get(ctx, cfgID)
	if err != nil {
		return err
	}
	if cfg.AccountType == model.AccountTypeFree {
		return errFreeAccountSkipped
	}
	end := time.Now().UTC()
	items, err := s.configs.Costs(ctx, cfgID, oci.CostQuery{
		StartTime: end.AddDate(0, 0, -7),
		EndTime:   end,
	})
	if err != nil {
		return err
	}
	return s.saveCostSnapshots(ctx, cfgID, items)
}

// saveCostSnapshots 把成本条目按 UTC 日聚合后逐日 upsert。
func (s *TaskService) saveCostSnapshots(ctx context.Context, cfgID uint, items []oci.CostItem) error {
	type bucket struct {
		amount   float64
		currency string
	}
	byDay := map[string]*bucket{}
	for _, item := range items {
		if item.TimeStart == nil {
			continue
		}
		day := item.TimeStart.UTC().Format("2006-01-02")
		b, ok := byDay[day]
		if !ok {
			b = &bucket{currency: item.Currency}
			byDay[day] = b
		}
		b.amount += float64(item.ComputedAmount)
	}
	now := time.Now()
	for day, b := range byDay {
		snap := model.CostSnapshot{
			OciConfigID: cfgID, Day: day,
			Amount: b.amount, Currency: b.currency, SyncedAt: now,
		}
		err := s.db.WithContext(ctx).Clauses(clause.OnConflict{
			Columns:   []clause.Column{{Name: "oci_config_id"}, {Name: "day"}},
			DoUpdates: clause.AssignmentColumns([]string{"amount", "currency", "synced_at"}),
		}).Create(&snap).Error
		if err != nil {
			return fmt.Errorf("save cost snapshot %s: %w", day, err)
		}
	}
	return nil
}

// targetConfigIDs 解析任务作用范围；未指定时返回全部配置 ID。
func (s *TaskService) targetConfigIDs(ctx context.Context, ids []uint) ([]uint, error) {
	if len(ids) > 0 {
		return ids, nil
	}
	configs, err := s.configs.List(ctx)
	if err != nil {
		return nil, err
	}
	all := make([]uint, 0, len(configs))
	for _, cfg := range configs {
		all = append(all, cfg.ID)
	}
	return all, nil
}

// runSnatch 尝试创建实例；抢到目标台数后任务标记 succeeded 并停止调度，
// 部分成功把剩余台数写回 payload 下次继续；成功路径一并清零并写回连续
// 鉴权失败计数，失败路径交给 snatchFailure 做连续 NotAuthenticated 熔断
// 判定。字段落库由 execute 统一 Save。
func (s *TaskService) runSnatch(ctx context.Context, task *model.Task) (string, error) {
	var p snatchPayload
	if err := json.Unmarshal([]byte(task.Payload), &p); err != nil {
		return "", fmt.Errorf("parse payload: %w", err)
	}
	if p.Count < 1 {
		p.Count = 1
	}
	in, adNote, err := s.snatchInstanceInput(ctx, task, &p)
	if err != nil {
		return "", s.snatchFailure(ctx, task, &p, err)
	}
	instances, failures, err := s.configs.CreateInstances(ctx, p.OciConfigID, in, p.Count)
	if err == nil && len(instances) == 0 {
		err = fmt.Errorf("no instance created%s: %s", adNote, strings.Join(failures, "; "))
	}
	if err != nil {
		return "", s.snatchFailure(ctx, task, &p, err)
	}
	p.AuthFailCount = 0
	ids := make([]string, 0, len(instances))
	for _, in := range instances {
		ids = append(ids, in.ID)
	}
	remaining := p.Count - len(instances)
	if remaining > 0 {
		p.Count = remaining
		writeSnatchPayload(task, &p)
		return fmt.Sprintf("created %d (%s)%s, %d remaining", len(instances), strings.Join(ids, ","), adNote, remaining), nil
	}
	s.snatchSucceed(ctx, task, &p, in, instances)
	return fmt.Sprintf("created %d: %s%s", len(instances), strings.Join(ids, ","), adNote), nil
}

// snatchSucceed 处理抢满收尾:标记成功、停止调度,并组装成功通知变量暂存,
// 供 execute 发通知时合并进 snatch_success 事件。
func (s *TaskService) snatchSucceed(ctx context.Context, task *model.Task, p *snatchPayload, in oci.CreateInstanceInput, instances []oci.Instance) {
	task.Status = model.TaskStatusSucceeded
	writeSnatchPayload(task, p)
	s.unschedule(task.ID)
	s.storeSnatchVars(task.ID, s.snatchSuccessVars(ctx, p, in, instances))
}

// snatchSuccessVars 组装抢机成功通知的模板变量;单项查询失败以占位值回退,
// 绝不阻塞通知发送。密码只进通知变量,不落任务日志。
func (s *TaskService) snatchSuccessVars(ctx context.Context, p *snatchPayload, in oci.CreateInstanceInput, instances []oci.Instance) map[string]string {
	tenant, region := "-", in.Region
	if cfg, err := s.configs.Get(ctx, p.OciConfigID); err == nil {
		tenant = cfg.Alias
		if region == "" {
			region = cfg.Region
		}
	}
	ips, pwds := s.snatchInstanceNet(ctx, p.OciConfigID, region, in, instances)
	return map[string]string{
		"tenant":        tenant,
		"region":        regionLabel(region),
		"shape":         in.Shape,
		"spec":          specLabel(in),
		"image":         s.snatchImageLabel(ctx, p.OciConfigID, region, in),
		"ip":            ips,
		"root_password": pwds,
	}
}

// snatchInstanceNet 逐台等待公网 IPv4 并取回 root 密码;全部实例共享
// snatchIPWait 总预算,多台结果按创建顺序以「、」连接。
func (s *TaskService) snatchInstanceNet(ctx context.Context, cfgID uint, region string, in oci.CreateInstanceInput, instances []oci.Instance) (string, string) {
	deadline := time.Now().Add(s.snatchIPWait)
	ips := make([]string, 0, len(instances))
	pwds := make([]string, 0, len(instances))
	for _, inst := range instances {
		ips = append(ips, s.waitPublicIPv4(ctx, cfgID, region, inst.ID, deadline))
		pwds = append(pwds, rootPasswordOf(in, inst))
	}
	return strings.Join(ips, "、"), strings.Join(pwds, "、")
}

// waitPublicIPv4 轮询实例公网 IPv4 直至分配或超过 deadline;
// 超时回退内网 IPv4(标注内网),再无则「待分配」。
func (s *TaskService) waitPublicIPv4(ctx context.Context, cfgID uint, region, instanceID string, deadline time.Time) string {
	private := ""
	for {
		inst, err := s.configs.Instance(ctx, cfgID, region, instanceID)
		if err == nil {
			if inst.PublicIP != "" {
				return inst.PublicIP
			}
			if inst.PrivateIP != "" {
				private = inst.PrivateIP
			}
		}
		if time.Now().After(deadline) {
			break
		}
		select {
		case <-ctx.Done():
			return waitIPFallback(private)
		case <-time.After(snatchIPPollInterval):
		}
	}
	return waitIPFallback(private)
}

// waitIPFallback 是公网 IP 等待超时后的展示回退。
func waitIPFallback(private string) string {
	if private != "" {
		return private + "(内网)"
	}
	return "待分配"
}

// rootPasswordOf 取单台实例的 root 密码:随机生成模式从创建时写入的
// FreeformTags 回读(每台独立),显式密码模式各台相同;SSH 密钥模式无密码。
func rootPasswordOf(in oci.CreateInstanceInput, inst oci.Instance) string {
	if pwd := inst.FreeformTags["RootPassword"]; pwd != "" {
		return pwd
	}
	if in.RootPassword != "" {
		return in.RootPassword
	}
	return "—(SSH 密钥登录)"
}

// specLabel 拼 Flex 规格,如 4C/24G;固定规格 shape(无 ocpus 参数)为 -。
func specLabel(in oci.CreateInstanceInput) string {
	if in.Ocpus <= 0 {
		return "-"
	}
	return fmt.Sprintf("%gC/%gG", in.Ocpus, in.MemoryInGBs)
}

// regionLabel 把区域标识换成控制台风格别名展示,如「Japan Central (Osaka)」;
// 区域名与三字码均可匹配,表中未收录时回退原值。
func regionLabel(region string) string {
	if region == "" {
		return "-"
	}
	regions, err := oci.AllRegions()
	if err != nil {
		return region
	}
	for _, r := range regions {
		if strings.EqualFold(r.Name, region) || strings.EqualFold(r.Key, region) {
			return r.Alias
		}
	}
	return region
}

// snatchImageLabel 返回启动源展示名:镜像查显示名(失败回退 OCID),
// 引导卷启动源固定文案。
func (s *TaskService) snatchImageLabel(ctx context.Context, cfgID uint, region string, in oci.CreateInstanceInput) string {
	if in.BootVolumeID != "" {
		return "引导卷"
	}
	if in.ImageID == "" {
		return "-"
	}
	img, err := s.configs.Image(ctx, cfgID, region, in.ImageID)
	if err != nil || img.DisplayName == "" {
		return in.ImageID
	}
	return img.DisplayName
}

// storeSnatchVars 暂存抢机成功通知变量,由 takeSnatchVars 取走。
func (s *TaskService) storeSnatchVars(taskID uint, vars map[string]string) {
	s.snatchVarsMu.Lock()
	defer s.snatchVarsMu.Unlock()
	s.snatchVars[taskID] = vars
}

// takeSnatchVars 取走并清除暂存的成功通知变量(无则 nil)。
func (s *TaskService) takeSnatchVars(taskID uint) map[string]string {
	s.snatchVarsMu.Lock()
	defer s.snatchVarsMu.Unlock()
	vars := s.snatchVars[taskID]
	delete(s.snatchVars, taskID)
	return vars
}

// snatchInstanceInput 组装本次创建参数:可用域显式指定时原样使用;
// 留空(自动)时按执行序号轮询区域全部可用域——ad-1、ad-2、ad-3 依次循环,
// 分摊单可用域容量不足。附加说明串供执行日志展示本次所用可用域。
func (s *TaskService) snatchInstanceInput(ctx context.Context, task *model.Task, p *snatchPayload) (oci.CreateInstanceInput, string, error) {
	in := p.Instance
	if in.AvailabilityDomain != "" {
		return in, "", nil
	}
	ads, err := s.configs.AvailabilityDomains(ctx, p.OciConfigID, in.Region)
	if err != nil {
		return in, "", fmt.Errorf("list availability domains: %w", err)
	}
	if len(ads) == 0 {
		return in, "", fmt.Errorf("region has no availability domain")
	}
	// execute 在 run 之后才递增 RunCount,此处即 0 起的本次执行序号
	in.AvailabilityDomain = ads[task.RunCount%len(ads)]
	return in, " @ " + in.AvailabilityDomain, nil
}

// snatchFailure 处理抢机单次失败：错误含 NotAuthenticated 时累计连续计数，
// 达阈值把任务置 failed 并移出调度（熔断）；其他错误清零计数。计数写回 payload。
func (s *TaskService) snatchFailure(ctx context.Context, task *model.Task, p *snatchPayload, cause error) error {
	if !strings.Contains(cause.Error(), "NotAuthenticated") {
		p.AuthFailCount = 0
		writeSnatchPayload(task, p)
		return cause
	}
	p.AuthFailCount++
	writeSnatchPayload(task, p)
	if p.AuthFailCount < s.snatchAuthFailLimit(ctx) {
		return cause
	}
	task.Status = model.TaskStatusFailed
	s.unschedule(task.ID)
	return fmt.Errorf("连续 %d 次 NotAuthenticated，任务已熔断停止: %w", p.AuthFailCount, cause)
}

// snatchAuthFailLimit 读取抢机熔断阈值；settings 未注入或读取失败按默认值。
func (s *TaskService) snatchAuthFailLimit(ctx context.Context) int {
	if s.settings == nil {
		return defaultSnatchAuthFailLimit
	}
	view, err := s.settings.TaskSettings(ctx)
	if err != nil {
		return defaultSnatchAuthFailLimit
	}
	return view.SnatchAuthFailLimit
}

// writeSnatchPayload 把最新抢机参数序列化写回任务（execute 统一落库）。
func writeSnatchPayload(task *model.Task, p *snatchPayload) {
	if raw, err := json.Marshal(p); err == nil {
		task.Payload = string(raw)
	}
}

// appendLog 写入执行日志并裁剪超出保留数量的旧日志。
func (s *TaskService) appendLog(taskID uint, success bool, message string, elapsed time.Duration) *model.TaskLog {
	entry := &model.TaskLog{
		TaskID:     taskID,
		Success:    success,
		Message:    message,
		DurationMs: elapsed.Milliseconds(),
	}
	s.db.Create(entry)
	s.db.Where("task_id = ? AND id NOT IN (?)", taskID,
		s.db.Model(&model.TaskLog{}).Select("id").Where("task_id = ?", taskID).
			Order("id desc").Limit(taskLogKeep),
	).Delete(&model.TaskLog{})
	return entry
}

// taskSnapshot 是通知判定所需的任务状态切片（执行前后各取一份）。
type taskSnapshot struct {
	Name      string
	Status    string
	LastError string
	Message   string // 本次执行结果摘要，仅执行后快照填写
	// SnatchVars 是抢机成功通知的补充变量(租户/区域/实例明细),
	// 仅抢满那次执行后快照携带
	SnatchVars map[string]string
}

// notifyKind 是通知事件类型，与设置页「通知管理」开关一一对应。
type notifyKind string

const (
	notifyTaskFail      notifyKind = "task_fail"
	notifyTaskRecover   notifyKind = "task_recover"
	notifySnatchSuccess notifyKind = "snatch_success"
	notifyTenantDead    notifyKind = "tenant_dead"
	notifyTaskStop      notifyKind = "task_stop" // 任务熔断停止（抢机连续鉴权失败达阈值）
)

// notifyEvent 是一条待发送的通知:类型供开关过滤与模板选择,Vars 为模板变量。
type notifyEvent struct {
	Kind notifyKind
	Vars map[string]string
}

// notifyEvents 比较执行前后的任务状态，返回需要推送的通知事件。
// 只在状态发生变化时产出：连续失败或持续正常都不重复发，防轰炸。
// 熔断翻转（置 failed）优先判定，该次只发任务停止、不叠加任务失败。
func notifyEvents(prev, cur taskSnapshot) []notifyEvent {
	var events []notifyEvent
	if prev.Status != model.TaskStatusSucceeded && cur.Status == model.TaskStatusSucceeded {
		vars := map[string]string{"task_name": cur.Name, "message": cur.Message}
		for k, v := range cur.SnatchVars {
			vars[k] = v
		}
		events = append(events, notifyEvent{notifySnatchSuccess, vars})
	}
	switch {
	case prev.Status != model.TaskStatusFailed && cur.Status == model.TaskStatusFailed:
		events = append(events, notifyEvent{notifyTaskStop, map[string]string{"task_name": cur.Name, "error": cur.LastError}})
	case prev.LastError == "" && cur.LastError != "":
		events = append(events, notifyEvent{notifyTaskFail, map[string]string{"task_name": cur.Name, "error": cur.LastError}})
	case prev.LastError != "" && cur.LastError == "" && cur.Status != model.TaskStatusSucceeded:
		// 恢复即成功收尾（抢机达成目标）时已有抢机成功通知，不再叠加恢复通知
		events = append(events, notifyEvent{notifyTaskRecover, map[string]string{"task_name": cur.Name}})
	}
	return events
}

// notifyFilterTimeout 是发送前查询事件开关的超时时间（本地 SQLite，查询极快）。
const notifyFilterTimeout = 5 * time.Second

// notify 逐条按事件开关过滤后异步发送；notifier 未注入（nil）时整体关闭。
// 开关读取失败按开启降级（NotifyEventEnabled 内部兜底），不因设置异常漏发。
func (s *TaskService) notify(events []notifyEvent) {
	if s.notifier == nil || len(events) == 0 {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), notifyFilterTimeout)
	defer cancel()
	for _, ev := range events {
		if s.settings != nil && !s.settings.NotifyEventEnabled(ctx, string(ev.Kind)) {
			continue
		}
		s.notifier.SendTemplateAsync(string(ev.Kind), ev.Vars)
	}
}

// deadAliasKey 是测活任务留存上次失联别名集合的 Setting 键。
func deadAliasKey(taskID uint) string {
	return fmt.Sprintf("health_dead_alias:%d", taskID)
}

// notifyDeadAliases 只在失联集合发生变化时推送失联通知，并留存本次集合；
// 集合不变（含持续失联）不重复发。notifier 未注入时整体跳过。
func (s *TaskService) notifyDeadAliases(ctx context.Context, taskID uint, aliases []string) {
	if s.notifier == nil {
		return
	}
	if sameStringSet(s.loadDeadAliases(ctx, taskID), aliases) {
		return
	}
	s.saveDeadAliases(ctx, taskID, aliases)
	if len(aliases) > 0 {
		s.notify([]notifyEvent{{notifyTenantDead, map[string]string{"tenants": strings.Join(aliases, "、")}}})
	}
}

// loadDeadAliases 读取任务上次记录的失联别名集合；无记录视为空集。
func (s *TaskService) loadDeadAliases(ctx context.Context, taskID uint) []string {
	var st model.Setting
	err := s.db.WithContext(ctx).First(&st, "key = ?", deadAliasKey(taskID)).Error
	if err != nil || st.Value == "" {
		return nil
	}
	var aliases []string
	if err := json.Unmarshal([]byte(st.Value), &aliases); err != nil {
		return nil
	}
	return aliases
}

// saveDeadAliases 覆盖保存本次失联别名集合；留存失败不影响任务执行。
func (s *TaskService) saveDeadAliases(ctx context.Context, taskID uint, aliases []string) {
	raw, err := json.Marshal(aliases)
	if err != nil {
		return
	}
	s.db.WithContext(ctx).Save(&model.Setting{
		Key: deadAliasKey(taskID), Value: string(raw), UpdatedAt: time.Now(),
	})
}

// sameStringSet 判断两个字符串切片内容是否相同（忽略顺序，重复元素按次数计）。
func sameStringSet(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	count := make(map[string]int, len(a))
	for _, v := range a {
		count[v]++
	}
	for _, v := range b {
		count[v]--
		if count[v] < 0 {
			return false
		}
	}
	return true
}