jh-zz/actions/sync.go

package actions

import (
	"fmt"
	"jhman/clients/jd"
	"jhman/model"
	"log/slog"

	"gorm.io/gorm"
)

// Sync 同步城市节点数据
//
// 数据库中保留云端已经移除的节点是有必要的，因为在节点轮换时需要保证能够从正确的位置开始，
// 而其之前所用的节点可能已经被云端移除，因此需要保留这些节点的记录以保持正确的轮换顺位。
func Sync() (err error) {
	slog.Info("开始同步城市节点数据")

	// 获取需要更新的城市列表
	cities, err := FindCities(model.DB)
	if err != nil {
		return fmt.Errorf("获取所有城市失败: %w", err)
	}
	slog.Info(fmt.Sprintf("成功获取城市列表，城市数量：%d", len(cities)))

	// 获取所有城市的节点数据
	for i, city := range cities {
		slog.Debug(fmt.Sprintf("正在同步城市，城市：%s，哈希：%s", city.Name, city.Hash))

		// 新节点信息
		resp, err := jd.EdgeDevice(jd.EdgeDeviceReq{
			Geo:    city.Hash,
			Offset: 0,
			Num:    1000000,
		})
		if err != nil {
			return fmt.Errorf("获取城市 %s:%s 的边缘设备失败: %w", city.Name, city.Hash, err)
		}
		var newEdges = resp.Edges
		slog.Debug(fmt.Sprintf("获取节点数据完成，城市：%s，节点数：%d", city.Name, len(newEdges)))

		err = model.DB.Transaction(func(tx *gorm.DB) error {
			// 旧节点信息
			oldEdges, err := FindEdgesByCity(tx, city.Id)
			if err != nil {
				return fmt.Errorf("获取所有边缘节点MAC地址失败: %w", err)
			}
			var oldEdgesMacSet = make(map[string]struct{}, len(oldEdges))
			for _, edge := range oldEdges {
				oldEdgesMacSet[edge.Macaddr] = struct{}{}
			}

			// 对比并更新节点信息（全量更新）
			var edgeSaves = make([]model.Edge, 0)
			for _, edge := range newEdges {
				edgeSaves = append(edgeSaves, model.Edge{
					CityId:  city.Id,
					Macaddr: edge.Macaddr,
					Public:  edge.Public,
					Isp:     edge.Isp,
					Single:  edge.Single,
					Sole:    edge.Sole,
					Arch:    edge.Arch,
					Online:  edge.Online,
					Active:  true,
				})
				delete(oldEdgesMacSet, edge.Macaddr)
			}

			// 更新现有数据
			if err = SaveEdges(tx, edgeSaves); err != nil {
				return fmt.Errorf("批量保存边缘节点失败: %w", err)
			}
			// 删除旧节点信息
			var oldEdgesMacs = make([]string, 0, len(oldEdgesMacSet))
			for mac := range oldEdgesMacSet {
				oldEdgesMacs = append(oldEdgesMacs, mac)
			}
			if err = DisableEdgesByMacs(tx, oldEdgesMacs); err != nil {
				return fmt.Errorf("通过MAC地址删除边缘节点失败: %w", err)
			}

			slog.Info(fmt.Sprintf("城市同步完成，新节点：%d，旧节点：%d，移除：%d，同步：%d，进度：%d/%d", len(newEdges), len(oldEdges), len(oldEdgesMacs), len(edgeSaves), i+1, len(cities)))
			return nil
		})
		if err != nil {
			return fmt.Errorf("城市 %s 事务处理失败: %w", city.Name, err)
		}

	}

	return nil
}