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

package tools

import (
	"context"
	"fmt"
	"time"

	"github.com/mark3labs/mcp-go/mcp"

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

// 站点信息表同步（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 {
	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）时任何情况下都不会重复触发"`
}

var SiteInfoRefresh = kit.Def[siteInfoRefreshArgs]{
	Name:  "site_info_refresh",
	Title: "触发站点信息表同步",
	Description: "触发 ToolBox 站点信息表从 FWH 平台重新同步，立即返回，同步在主服务后台执行，耗时可能超过 1.5 小时。" +
		"触发前先检查同步状态：同步进行中时不重复触发，返回当前进度（action=tracking）；最近 24 小时内已成功同步时不触发（action=skipped_fresh）；" +
		"上次同步超过 6 小时未完成、疑似中断时不触发（action=skipped_stale）；以上两种不触发的情况可用 force=true 强制触发。" +
		"已触发时 action=triggered。返回中的 status 与 site_info_sync_status 相同，之后用 site_info_sync_status 跟踪进度。" +
		"建议先调用 site_info_sync_status 判断是否需要同步，再调用本工具。这是写操作，执行前会请求确认。",
	ReadOnly: false,
	// 同步中或数据较新时不重复触发，网络出错时平台重试一次不会产生第二次同步
	Idempotent: true,
	Handler: func(ctx context.Context, a siteInfoRefreshArgs) (*mcp.CallToolResult, error) {
		s, err := loadSiteInfoSync()
		if err != nil {
			return kit.Upstream("ToolBox 站点信息同步记录", err), nil
		}
		action, trigger := s.decideRefresh(a.Force)
		if trigger {
			if err := services.UpdateSiteInfoDate(); err != nil {
				return kit.Upstream("ToolBox 站点信息同步", err), nil
			}
			if s, err = loadSiteInfoSync(); err != nil {
				return kit.Upstream("ToolBox 站点信息同步记录", err), nil
			}
		}
		return kit.Object(map[string]any{"action": action, "message": refreshMessages[action], "status": s.view()}), nil
	},
}