FranklinWH-Toolbox/service/agentmcp/tools/siteinfosync.go

package tools

import (
	"context"
	"fmt"
	"time"

	"github.com/mark3labs/mcp-go/mcp"
	"go.mongodb.org/mongo-driver/bson/primitive"

	"fwh-toolbox-server/agentmcp/kit"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/repositories"
	"fwh-toolbox-server/services"
	"fwh-toolbox-server/task/queue/fwhQueue"
)

// 站点信息表同步（SiteInfoManageRecord）的状态判断。
// 同步由主服务的任务队列执行，耗时可能超过 1.5 小时；记录只有 progress、createTime（请求时间）、
// updateTime（完成时间，仅 complete 时更新）与 total，没有心跳，只能按请求时间判断是否疑似中断。
const (
	// syncStaleAfter pending / running 超过该时长视为疑似中断（例如主服务重启），需要 force 才重新触发
	syncStaleAfter = 6 * time.Hour
	// syncFreshWithin 最近一次成功同步在该时长内时认为数据较新，不重新触发（force 除外）
	syncFreshWithin = 24 * time.Hour
)

// 同步状态
const (
	syncStateNever   = "never"   // 从未同步
	syncStateSyncing = "syncing" // 同步中（pending / running）
	syncStateStale   = "stale"   // pending / running 但超过 syncStaleAfter，疑似中断
	syncStateIdle    = "idle"    // 没有进行中的同步（最近一次已完成或失败）
)

var beijing = time.FixedZone("CST", 8*3600)

func fmtBeijing(t time.Time) string {
	return t.In(beijing).Format(time.DateTime)
}

func fmtHours(d time.Duration) string {
	if d < time.Hour {
		return fmt.Sprintf("%d 分钟", int(d.Minutes()))
	}
	return fmt.Sprintf("%.1f 小时", d.Hours())
}

type siteInfoSync struct {
	State       string
	Latest      *domain.SiteInfoUpdateRecord
	LastSuccess *domain.SiteInfoUpdateRecord
	now         time.Time
}

func loadSiteInfoSync() (*siteInfoSync, error) {
	latest, err := repositories.GetSiteInfoUpdateNewRecord()
	if err != nil {
		return nil, err
	}
	lastComplete, err := repositories.GetSiteInfoUpdateLastComplete()
	if err != nil {
		return nil, err
	}
	return evalSiteInfoSync(latest, lastComplete, time.Now()), nil
}

// evalSiteInfoSync 根据最近一次记录与最近一次成功记录判断同步状态
func evalSiteInfoSync(latest, lastComplete *domain.SiteInfoUpdateRecord, now time.Time) *siteInfoSync {
	// 最近一条已完成时它就是最近一次成功（调用方可能只传了最近一条）
	if lastComplete == nil && latest != nil && latest.Progress == domain.ProgressComplete {
		lastComplete = latest
	}
	s := &siteInfoSync{Latest: latest, LastSuccess: lastComplete, now: now}
	switch {
	case latest == nil:
		s.State = syncStateNever
	case latest.Progress == domain.ProgressPending || latest.Progress == domain.ProgressRunning:
		s.State = syncStateSyncing
		if s.now.Sub(latest.CreateTime) > syncStaleAfter {
			s.State = syncStateStale
		}
	default:
		s.State = syncStateIdle
	}
	return s
}

// decideRefresh 决定 site_info_refresh 是否触发同步；trigger=false 时 action 说明原因
func (s *siteInfoSync) decideRefresh(force bool) (action string, trigger bool) {
	switch {
	case s.State == syncStateSyncing:
		return "tracking", false
	case s.State == syncStateStale && !force:
		return "skipped_stale", false
	case s.State == syncStateIdle && s.Latest.Progress == domain.ProgressComplete && s.fresh() && !force:
		return "skipped_fresh", false
	}
	return "triggered", true
}

// fresh 最近一次成功同步在 syncFreshWithin 内
func (s *siteInfoSync) fresh() bool {
	return s.LastSuccess != nil && s.now.Sub(s.LastSuccess.UpdateTime) < syncFreshWithin
}

func (s *siteInfoSync) view() map[string]any {
	out := map[string]any{"state": s.State}
	if r := s.Latest; r != nil {
		latest := map[string]any{
			"id":          r.Id.Hex(),
			"progress":    r.Progress,
			"requestedAt": fmtBeijing(r.CreateTime),
		}
		if r.Progress == domain.ProgressComplete {
			latest["completedAt"] = fmtBeijing(r.UpdateTime)
			latest["duration"] = fmtHours(r.UpdateTime.Sub(r.CreateTime))
			latest["total"] = r.Total
		} else {
			latest["elapsed"] = fmtHours(s.now.Sub(r.CreateTime))
		}
		out["latest"] = latest
	}
	if r := s.LastSuccess; r != nil {
		out["lastSuccess"] = map[string]any{
			"completedAt": fmtBeijing(r.UpdateTime),
			"age":         fmtHours(s.now.Sub(r.UpdateTime)),
			"total":       r.Total,
		}
	}
	out["advice"] = s.advice()
	return out
}

func (s *siteInfoSync) advice() string {
	switch s.State {
	case syncStateNever:
		return "从未同步过，可以调用 site_info_refresh 触发同步。"
	case syncStateSyncing:
		return fmt.Sprintf("同步进行中（已 %s），无需重新触发。同步耗时可能超过 1.5 小时，稍后再用 site_info_sync_status 查看。",
			fmtHours(s.now.Sub(s.Latest.CreateTime)))
	case syncStateStale:
		return fmt.Sprintf("最近一次同步已请求 %s 仍未完成，超过 %s，疑似中断（如主服务重启）。确认后可用 site_info_refresh 并设置 force=true 重新同步。",
			fmtHours(s.now.Sub(s.Latest.CreateTime)), fmtHours(syncStaleAfter))
	}
	if s.Latest.Progress != domain.ProgressComplete {
		return "最近一次同步失败，可以调用 site_info_refresh 重新同步。"
	}
	if s.fresh() {
		return fmt.Sprintf("最近一次同步在 %s 前完成，数据较新，一般无需重新同步；确需最新数据时用 site_info_refresh 并设置 force=true。",
			fmtHours(s.now.Sub(s.LastSuccess.UpdateTime)))
	}
	return fmt.Sprintf("最近一次同步在 %s 前完成，超过 %s，可以调用 site_info_refresh 重新同步。",
		fmtHours(s.now.Sub(s.LastSuccess.UpdateTime)), fmtHours(syncFreshWithin))
}

// ---- site_info_sync_status ----

var SiteInfoSyncStatus = kit.Def[kit.NoArgs]{
	Name:  "site_info_sync_status",
	Title: "查询站点信息表同步状态",
	Description: "查询 ToolBox 站点信息表（site_search 的数据来源）从 FWH 平台同步的状态，只读。" +
		"返回 state：never（从未同步）、syncing（同步中）、stale（超过 6 小时未完成，疑似中断）、idle（没有进行中的同步）；" +
		"latest（最近一次同步：progress 为 pending/running/complete/error，requestedAt 请求时间，完成时有 completedAt、duration、total 站点数，未完成时有 elapsed）；" +
		"lastSuccess（最近一次成功同步的 completedAt、age 距今时长、total）；advice（是否需要重新同步的建议）。时间为北京时间。" +
		"同步耗时可能超过 1.5 小时：触发同步前先调用本工具，同步中时用本工具跟踪进度，不要重复触发。",
	ReadOnly: true,
	Handler: func(ctx context.Context, _ kit.NoArgs) (*mcp.CallToolResult, error) {
		s, err := loadSiteInfoSync()
		if err != nil {
			return kit.Upstream("ToolBox 站点信息同步记录", err), nil
		}
		return kit.Object(s.view()), nil
	},
}

// ---- site_info_refresh ----

var refreshMessages = map[string]string{
	"tracking":      "同步进行中，未重复触发。",
	"skipped_stale": "上次同步疑似中断，未触发；确认后设置 force=true 重新同步。",
	"skipped_fresh": "最近 24 小时内已成功同步，未触发；确需最新数据时设置 force=true。",
	"triggered":     "已触发同步，任务在主服务后台执行，耗时可能超过 1.5 小时；用 site_info_sync_status 跟踪进度。",
}

type siteInfoRefreshArgs struct {
	Force bool `json:"force,omitempty" jsonschema:"为 true 时：最近 24 小时内已成功同步、或上次同步疑似中断（stale）时仍重新触发。同步进行中（syncing）时任何情况下都不会重复触发"`
}

// siteInfoTaskPoll 平台查询任务状态的间隔：同步约 1 小时，每分钟查一次足够
const siteInfoTaskPoll = time.Minute

var SiteInfoRefresh = kit.TaskDef[siteInfoRefreshArgs]{
	Def: kit.Def[siteInfoRefreshArgs]{
		Name:  "site_info_refresh",
		Title: "触发站点信息表同步",
		Description: "触发 ToolBox 站点信息表从 FWH 平台重新同步，同步在主服务后台执行，耗时约 1 小时，可能超过 1.5 小时。" +
			"触发前会检查同步状态：同步进行中时不重复触发，改为跟踪这次同步（action=tracking）；最近 24 小时内已成功同步时不触发（action=skipped_fresh）；" +
			"上次同步超过 6 小时未完成、疑似中断时不触发（action=skipped_stale）；这两种不触发的情况可用 force=true 强制触发。已触发时 action=triggered。" +
			"以长任务方式执行：触发或跟踪时立即返回“已转为后台”，同步结束后平台会把结果（完成时间、站点数、耗时，或失败原因）发给你，无需轮询；" +
			"用户询问进度时用 site_info_sync_status。这是写操作，执行前会请求确认。",
		ReadOnly: false,
		// 同步中或数据较新时不重复触发，网络出错时平台重试一次不会产生第二次同步
		Idempotent: true,
		Handler: func(ctx context.Context, a siteInfoRefreshArgs) (*mcp.CallToolResult, error) {
			action, s, fail := refreshSiteInfo(a.Force)
			if fail != nil {
				return fail, nil
			}
			return kit.Object(map[string]any{"action": action, "message": refreshMessages[action], "status": s.view()}), nil
		},
	},
	Poll: siteInfoTaskPoll,
	Start: func(ctx context.Context, a siteInfoRefreshArgs) (kit.TaskStart, *mcp.CallToolResult) {
		action, s, fail := refreshSiteInfo(a.Force)
		if fail != nil {
			return kit.TaskStart{}, fail
		}
		// 每种结果都对应一条同步记录：触发的是新记录，跟踪、不触发的是最近一条记录
		note := refreshMessages[action]
		if action == "triggered" || action == "tracking" {
			note += "同步结束后平台会把结果发给你，现在可以先告诉用户已提交。"
		}
		return kit.TaskStart{Key: s.Latest.Id.Hex(), Flag: action, Note: note}, nil
	},
	Status: func(ctx context.Context, ref kit.TaskRef) (kit.TaskState, error) {
		id, err := primitive.ObjectIDFromHex(ref.Key)
		if err != nil {
			return kit.TaskState{}, kit.ErrTaskNotFound
		}
		rec, err := repositories.GetSiteInfoUpdateRecord(id)
		if err != nil {
			return kit.TaskState{}, err
		}
		if rec == nil {
			return kit.TaskState{}, kit.ErrTaskNotFound
		}
		return siteInfoTaskState(ref.Flag, rec, time.Now()), nil
	},
	Cancel: func(ctx context.Context, ref kit.TaskRef) error {
		// 只有本次调用触发的同步才真正停止；跟踪他人触发的同步时只是停止跟踪（其他会话可能还在等它）
		if ref.Flag != "triggered" {
			return nil
		}
		id, err := primitive.ObjectIDFromHex(ref.Key)
		if err != nil {
			return kit.ErrTaskNotFound
		}
		return fwhQueue.CancelSiteInfoUpdate(id)
	},
}

// refreshSiteInfo 判断并在需要时触发同步，返回动作与触发后的状态；普通调用与任务方式调用共用
func refreshSiteInfo(force bool) (string, *siteInfoSync, *mcp.CallToolResult) {
	s, err := loadSiteInfoSync()
	if err != nil {
		return "", nil, kit.Upstream("ToolBox 站点信息同步记录", err)
	}
	action, trigger := s.decideRefresh(force)
	if trigger {
		if err := services.UpdateSiteInfoDate(); err != nil {
			return "", nil, kit.Upstream("ToolBox 站点信息同步", err)
		}
		if s, err = loadSiteInfoSync(); err != nil {
			return "", nil, kit.Upstream("ToolBox 站点信息同步记录", err)
		}
	}
	return action, s, nil
}

// siteInfoTaskState 把同步记录翻译为任务状态。flag 为发起任务时的动作：
// skipped_* 时任务立即完成（结果是不触发的原因）；triggered / tracking 时跟随这条同步记录
func siteInfoTaskState(flag string, rec *domain.SiteInfoUpdateRecord, now time.Time) kit.TaskState {
	st := kit.TaskState{CreatedAt: rec.CreateTime}
	if flag == "skipped_fresh" || flag == "skipped_stale" {
		st.Status, st.Message = mcp.TaskStatusCompleted, "未触发同步"
		st.Result = kit.Object(map[string]any{"action": flag, "message": refreshMessages[flag], "status": evalSiteInfoSync(rec, nil, now).view()})
		return st
	}
	elapsed := now.Sub(rec.CreateTime)
	switch rec.Progress {
	case domain.ProgressPending, domain.ProgressRunning:
		if elapsed > syncStaleAfter {
			st.Status = mcp.TaskStatusFailed
			st.Message = fmt.Sprintf("同步已 %s 仍未完成，疑似中断", fmtHours(elapsed))
			st.Result = kit.Failf("站点信息同步已请求 %s 仍未完成，超过 %s，疑似中断（如主服务重启）。可以告诉用户，确认后用 site_info_refresh 并设置 force=true 重新同步。",
				fmtHours(elapsed), fmtHours(syncStaleAfter))
			return st
		}
		st.Status = mcp.TaskStatusWorking
		if rec.Progress == domain.ProgressPending {
			st.Message = fmt.Sprintf("排队中，已等待 %s", fmtHours(elapsed))
		} else {
			st.Message = fmt.Sprintf("同步中，已进行 %s", fmtHours(elapsed))
		}
	case domain.ProgressComplete:
		st.Status, st.Message = mcp.TaskStatusCompleted, "同步完成"
		st.Result = kit.Object(map[string]any{
			"message":     fmt.Sprintf("站点信息表同步完成：%d 个站点，耗时 %s，完成于 %s（北京时间）。", rec.Total, fmtHours(rec.UpdateTime.Sub(rec.CreateTime)), fmtBeijing(rec.UpdateTime)),
			"requestedAt": fmtBeijing(rec.CreateTime),
			"completedAt": fmtBeijing(rec.UpdateTime),
			"total":       rec.Total,
		})
	case domain.ProgressCancelled:
		st.Status, st.Message = mcp.TaskStatusCancelled, "同步已取消"
		st.Result = kit.Failf("站点信息同步已取消。")
	default: // error / failed
		st.Status, st.Message = mcp.TaskStatusFailed, "同步失败"
		st.Result = kit.Failf("站点信息同步失败（主服务同步任务出错，详情见主服务日志）。可以稍后用 site_info_refresh 重新同步。")
	}
	return st
}