// +======================================================================
// | EasyGoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ]
// +----------------------------------------------------------------------
// | Copyright (c) 2019-2026 半城风雨团队 版权所有
// +----------------------------------------------------------------------
// | 开源协议: Apache-2.0
// | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。
// +----------------------------------------------------------------------
// | 官方网站: https://www.easygoadmin.vip
// +----------------------------------------------------------------------
// | 研发团队: 小蚂蚁云团队 荣誉出品
// +======================================================================
// | 版权和免责声明:
// | 一、知识产权声明
// | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、
// | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权，
// | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。
// | 二、使用限制与禁止条款
// | 您理解并同意，禁止将本产品用于以下任何行为：
// | 1. 任何违反中华人民共和国法律法规的用途；
// | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为；
// | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。
// | 三、责任豁免
// | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失
// | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三
// | 方提起索赔或行政机关进行调查，您应独立承担全部责任。
// | 四、违约与法律责任
// | 如发现任何违反本声明的行为，本团队有权：
// | 1. 无条件配合公安机关及其他有权机关的调查取证工作；
// | 2. 依法追究相关方的民事责任，并要求赔偿全部损失；
// | 3. 保留进一步采取刑事控告等法律措施的权利。
// | 五、许可范围说明
// | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包
// | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通
// | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、
// | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造
// | 成的一切经济损失及商誉损害。
// | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件
// | 许可协议》附件为准。
// +======================================================================

package logic

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/net/gclient"
	"github.com/gogf/gf/v2/os/gcron"

	"easygoadmin/internal/dao"
	"easygoadmin/utility/cache"
)

// ============================================================================
// 定时任务调度器：基于 gcron 的常驻调度，启动时加载 status=1 的任务，
// 按 cron_expression 触发 url+params（HTTP 调用本服务接口），结果写任务日志表。
//
// 任务键：job:{id}（gcron 命名条目），暂停=Remove、恢复=重新 Add。
// ============================================================================

var (
	daemonOnce  sync.Once // 保证调度器仅初始化一次
	daemonStart time.Time // 调度器启动时间
)

// jobLockTTLSeconds 任务执行锁 TTL（秒）
// 兜底上限：执行实例崩溃未主动释放时锁自动过期；正常路径执行结束即释放
const jobLockTTLSeconds = 300

// jobTryLock 抢占任务执行锁
//
// 键 job:lock:{id} 经 utility/cache（Redis 优先、内存降级）SET NX 实现，
// 多实例部署共享同一把锁，防止同一任务被并发执行；
// 抢占成功返回释放函数，失败返回 nil（调用方放弃本次执行）
func jobTryLock(jobId int64) func() {
	key := fmt.Sprintf("job:lock:%d", jobId)
	ctx := context.Background()
	if !cache.SetNX(ctx, key, "1", jobLockTTLSeconds) {
		return nil
	}
	return func() { _ = cache.Del(ctx, key) }
}

// jobKey 生成 gcron 条目键
func jobKey(jobId int64) string {
	return fmt.Sprintf("job:%d", jobId)
}

// DaemonInit 启动时加载运行中的任务
// 幂等，多次调用仅首次生效
func DaemonInit(ctx context.Context) {
	daemonOnce.Do(func() {
		daemonStart = time.Now()
		rows, err := dao.Job.Ctx(ctx).Where("status", 1).All()
		if err != nil {
			g.Log().Warningf(ctx, "定时任务加载失败: %v", err)
			return
		}
		g.Log().Debugf(ctx, "定时任务待挂载: %d 条", len(rows))
		loaded := 0
		for _, r := range rows {
			if err := addJobToScheduler(r["id"].Int64(), r["cron_expression"].String(),
				r["execute_policy"].Int(), r["is_sync"].Int()); err != nil {
				g.Log().Warningf(ctx, "任务挂载失败 job=%d cron=%q: %v",
					r["id"].Int64(), r["cron_expression"].String(), err)
				continue
			}
			loaded++
		}
		g.Log().Infof(ctx, "定时任务调度器已启动，加载 %d 个任务", loaded)
	})
}

// addJobToScheduler 挂载任务到调度器
//
// 执行策略（execute_policy）：
//   - 1 立即执行：允许并发，tick 即启动新执行（gcron.Add）
//   - 2 执行一次（默认）：单例调度，上次未结束则跳过本次（gcron.AddSingleton）
//   - 3 放弃执行：并发 tick 照常触发，但抢不到执行锁则放弃本次
//
// 执行方式（is_sync）：1 同步在调度协程内执行；0 异步 go 协程执行。
// 策略 2/3 均叠加执行锁（多实例共享），策略 1 按并发语义不加锁。
func addJobToScheduler(jobId int64, cronExpr string, executePolicy int, isSync int) error {
	if cronExpr == "" {
		return fmt.Errorf("cron表达式为空")
	}
	handler := func(c context.Context) {
		run := func(ctx context.Context) {
			if executePolicy != 1 {
				unlock := jobTryLock(jobId)
				if unlock == nil {
					g.Log().Debugf(ctx, "任务 %d 存在执行中的实例，本次放弃", jobId)
					return
				}
				defer unlock()
			}
			executeJob(ctx, jobId, false)
		}
		if isSync == 1 {
			run(c)
		} else {
			go run(context.Background())
		}
	}
	var err error
	if executePolicy == 1 {
		_, err = gcron.Add(context.Background(), cronExpr, handler, jobKey(jobId))
	} else {
		_, err = gcron.AddSingleton(context.Background(), cronExpr, handler, jobKey(jobId))
	}
	return err
}

// removeJobFromScheduler 从调度器摘除任务
func removeJobFromScheduler(jobId int64) {
	gcron.Remove(jobKey(jobId))
}

// hasJobInScheduler 判断调度器中是否包含指定任务
func hasJobInScheduler(jobId int64) bool {
	return gcron.Search(jobKey(jobId)) != nil
}

// executeJob 执行任务
//
// 流程：读取任务定义 → HTTP 调用目标地址 → 写入执行日志；
// 内部捕获 panic，避免异常外泄影响调度器。
// force 为 true 时跳过状态检查，用于手动"立即执行"场景，未发布任务也可执行。
func executeJob(ctx context.Context, jobId int64, force bool) {
	defer func() {
		if e := recover(); e != nil {
			g.Log().Errorf(ctx, "定时任务 %d 执行 panic: %v", jobId, e)
		}
	}()
	one, err := dao.Job.Ctx(ctx).Where("id", jobId).One()
	if err != nil || len(one) == 0 {
		removeJobFromScheduler(jobId)
		return
	}
	if !force && one["status"].Int() != 1 {
		removeJobFromScheduler(jobId)
		return
	}

	start := time.Now()
	status := 0
	message := "执行成功"
	exception := ""

	// 任务执行目标分派：
	//   http(s) 地址 → HTTP 调用；其余取值 → 按方法名查找本地处理方法执行
	target := one["url"].String()
	if target == "" {
		status = 1
		message = "执行失败"
		exception = "任务URL为空"
	} else if strings.HasPrefix(target, "http://") || strings.HasPrefix(target, "https://") {
		client := gclient.New()
		client.SetContentType("application/json")
		client.SetTimeout(30 * time.Second)
		body := one["params"].String()
		var response *gclient.Response
		var rerr error
		if body == "" {
			response, rerr = client.Get(ctx, target)
		} else {
			response, rerr = client.Post(ctx, target, body)
		}
		if rerr != nil {
			status = 1
			message = "执行失败"
			exception = truncStr(rerr.Error(), 1000)
		} else {
			respBody := truncStr(response.ReadAllString(), 1000)
			if response.StatusCode != 200 {
				status = 1
				message = "执行失败"
				exception = fmt.Sprintf("HTTP %d: %s", response.StatusCode, respBody)
			} else {
				message = truncStr("执行成功: "+respBody, 500)
			}
		}
	} else {
		handler, ok := jobHandlers[target]
		if !ok {
			status = 1
			message = "执行失败"
			exception = fmt.Sprintf("未找到本地任务处理方法: %s", truncStr(target, 200))
		} else {
			summary, herr := handler(ctx, one["params"].String())
			if herr != nil {
				status = 1
				message = "执行失败"
				exception = truncStr(herr.Error(), 1000)
			} else if summary != "" {
				message = truncStr("执行成功: "+summary, 500)
			}
		}
	}
	end := time.Now()

	// 执行日志落库（后台 ctx，独立于任务 ctx）
	if _, werr := dao.JobLog.Ctx(ctx).Data(map[string]any{
		"job_id":          jobId,
		"job_name":        one["job_name"].String(),
		"job_group":       one["job_group"].String(),
		"job_trigger":     one["job_trigger"].String(),
		"job_message":     message,
		"cron_expression": one["cron_expression"].String(),
		"status":          status,
		"start_time":      start,
		"end_time":        end,
		"consume_time":    end.Sub(start).Milliseconds(),
		"exception_info":  exception,
		"create_user":     "daemon",
	}).Insert(); werr != nil {
		g.Log().Errorf(ctx, "任务日志写入失败 job=%d: %v", jobId, werr)
	}
}

// truncStr 按字符数截断字符串
// 长度未超限时原样返回，超限时按 rune 截断，避免截断多字节字符
func truncStr(s string, maxLen int) string {
	if len(s) <= maxLen {
		return s
	}
	runes := []rune(s)
	if len(runes) <= maxLen {
		return s
	}
	return string(runes[:maxLen])
}