aiops/backend/controller/cloud_sync.go

package controller

import (
	"net/http"
	"strconv"
	"sync"
	"time"

	"ezops/cloud"
	"ezops/model"
	"ezops/pkg/crypto"

	"github.com/gin-gonic/gin"
	"gorm.io/gorm"
)

// CloudSyncHandler 云平台同步处理器
type CloudSyncHandler struct {
	db    *gorm.DB
	crypt *crypto.SecretBox
}

func NewCloudSyncHandler(db *gorm.DB, masterKey string) *CloudSyncHandler {
	return &CloudSyncHandler{db: db, crypt: crypto.NewSecretBox(masterKey)}
}

// SyncRegions 同步云账号的区域列表到本地 regions 表
// POST /cloud-accounts/:id/sync-regions
func (h *CloudSyncHandler) SyncRegions(c *gin.Context) {
	id, _ := strconv.Atoi(c.Param("id"))

	var account model.CloudAccount
	if err := h.db.First(&account, id).Error; err != nil {
		c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "云账号不存在"})
		return
	}
	if account.Status != "active" {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "云账号未激活，无法同步"})
		return
	}

	provider := cloud.NewProvider(account.Provider, account.AccessKey, account.SecretKey)
	if provider == nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "暂不支持的云平台: " + account.Provider})
		return
	}

	cloudRegions, err := provider.ListRegions()
	if err != nil {
		c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: "同步区域失败: " + err.Error()})
		return
	}

	// Upsert: 不存在则创建，存在则更新
	synced := 0
	for _, cr := range cloudRegions {
		var region model.Region
		result := h.db.Where("code = ? AND provider = ?", cr.RegionID, account.Provider).First(&region)
		if result.Error == gorm.ErrRecordNotFound {
			// 新区域 → 创建
			h.db.Create(&model.Region{
				Name:     cr.LocalName,
				Code:     cr.RegionID,
				Provider: account.Provider,
				Endpoint: cr.Endpoint,
				Status:   "active",
			})
			synced++
		} else if result.Error == nil {
			// 已存在 → 更新 endpoint 和名称
			region.Name = cr.LocalName
			region.Endpoint = cr.Endpoint
			h.db.Save(&region)
		}
	}

	// 并发统计所有可用区域实例数，卡片展示云账号真实资源量。 // Logan.Li
	serverCount, countedRegions := countCloudInstances(provider, cloudRegions)
	updates := map[string]interface{}{
		"region_count": len(cloudRegions),
		"last_sync_at": time.Now(),
	}
	if countedRegions > 0 {
		updates["server_count"] = serverCount
	}
	// 更新账号区域、实例数量和同步时间。 // Logan.Li
	h.db.Model(&account).Updates(updates)

	c.JSON(http.StatusOK, model.Response{
		Code:    200,
		Message: "同步成功",
		Data: gin.H{
			"synced":       synced,
			"total":        len(cloudRegions),
			"server_count": serverCount,
		},
	})
}

// countCloudInstances 以固定并发统计各区域实例数量，避免串行请求过慢。 // Logan.Li
func countCloudInstances(provider cloud.Provider, regions []cloud.CloudRegion) (total int, succeeded int) {
	type countResult struct {
		count int
		ok    bool
	}
	results := make(chan countResult, len(regions))
	semaphore := make(chan struct{}, 4)
	var waitGroup sync.WaitGroup
	for _, region := range regions {
		regionID := region.RegionID
		waitGroup.Add(1)
		go func() {
			defer waitGroup.Done()
			semaphore <- struct{}{}
			defer func() { <-semaphore }()
			instances, err := provider.ListInstances(regionID)
			results <- countResult{count: len(instances), ok: err == nil}
		}()
	}
	waitGroup.Wait()
	close(results)
	for result := range results {
		if result.ok {
			total += result.count
			succeeded++
		}
	}
	return total, succeeded
}

// GetAccountRegions 获取指定云账号的区域列表（按 provider 过滤）
// GET /cloud-accounts/:id/regions
func (h *CloudSyncHandler) GetAccountRegions(c *gin.Context) {
	id, _ := strconv.Atoi(c.Param("id"))

	var account model.CloudAccount
	if err := h.db.First(&account, id).Error; err != nil {
		c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "云账号不存在"})
		return
	}

	var regions []model.Region
	h.db.Where("provider = ?", account.Provider).Order("name ASC").Find(&regions)

	c.JSON(http.StatusOK, model.Response{
		Code:    200,
		Message: "success",
		Data: model.ListResponse{
			List:  regions,
			Total: int64(len(regions)),
		},
	})
}

// discoverRequest 发现主机请求
type discoverRequest struct {
	Regions []string `json:"regions"`
}

// DiscoverInstances 发现指定区域的云主机实例
// POST /cloud-accounts/:id/discover-instances
func (h *CloudSyncHandler) DiscoverInstances(c *gin.Context) {
	id, _ := strconv.Atoi(c.Param("id"))

	var account model.CloudAccount
	if err := h.db.First(&account, id).Error; err != nil {
		c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "云账号不存在"})
		return
	}

	var req discoverRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "请求参数错误"})
		return
	}
	if len(req.Regions) == 0 {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "请至少选择一个区域"})
		return
	}

	provider := cloud.NewProvider(account.Provider, account.AccessKey, account.SecretKey)
	if provider == nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "暂不支持的云平台"})
		return
	}

	var allInstances []cloud.CloudInstance
	var errors []string

	for _, regionID := range req.Regions {
		instances, err := provider.ListInstances(regionID)
		if err != nil {
			errors = append(errors, regionID+": "+err.Error())
			continue
		}
		allInstances = append(allInstances, instances...)
	}

	c.JSON(http.StatusOK, model.Response{
		Code:    200,
		Message: "success",
		Data: gin.H{
			"list":   allInstances,
			"total":  len(allInstances),
			"errors": errors,
		},
	})
}

// importRequest 导入主机请求
type importInstancesRequest struct {
	Instances       []cloud.CloudInstance `json:"instances"`
	DefaultPort     int                   `json:"default_port"`
	CredentialID    *uint                 `json:"credential_id"`     // 兼容旧字段：默认 SSH 凭据 // Logan.Li
	SSHCredentialID *uint                 `json:"ssh_credential_id"` // Linux 默认 SSH 凭据 // Logan.Li
	RDPCredentialID *uint                 `json:"rdp_credential_id"` // Windows 默认 RDP 凭据 // Logan.Li
}

// ImportInstances 将发现的云主机实例导入为本地服务器资产
// POST /cloud-accounts/:id/import-instances
func (h *CloudSyncHandler) ImportInstances(c *gin.Context) {
	id, _ := strconv.Atoi(c.Param("id"))

	var account model.CloudAccount
	if err := h.db.First(&account, id).Error; err != nil {
		c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "云账号不存在"})
		return
	}

	var req importInstancesRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "请求参数错误"})
		return
	}

	defaultPort := req.DefaultPort
	if defaultPort <= 0 {
		defaultPort = 22
	}

	sshCredentialID := req.SSHCredentialID
	if sshCredentialID == nil {
		sshCredentialID = req.CredentialID
	}
	sshCred, sshPlain, hasSSHCred, err := h.loadImportCredential(sshCredentialID, "ssh")
	if err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: err.Error()})
		return
	}
	rdpCred, rdpPlain, hasRDPCred, err := h.loadImportCredential(req.RDPCredentialID, "rdp")
	if err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: err.Error()})
		return
	}

	imported := 0
	skipped := 0

	for _, inst := range req.Instances {
		// 以 IP 去重（没有 IP 则用 InstanceID）
		ip := inst.PrivateIP
		if ip == "" {
			ip = inst.PublicIP
		}
		if ip == "" {
			ip = inst.InstanceID
		}

		// 重复资产补齐来源账号关联，保证服务器数统计准确。 // Logan.Li
		var existing model.Server
		if err := h.db.Where("ip = ? OR hostname = ?", ip, inst.Hostname).Take(&existing).Error; err == nil {
			if existing.CloudAccountID == nil {
				h.db.Model(&existing).Update("cloud_account_id", account.ID)
			}
			skipped++
			continue
		}

		// 查找 Region 关联
		var region model.Region
		var regionID *uint
		if h.db.Where("code = ? AND provider = ?", inst.RegionID, account.Provider).First(&region).Error == nil {
			regionID = &region.ID
		}

		osName := inst.OSType
		osType := normalizeImportOSType(inst.OSType, inst.InstanceName)
		protocol := "ssh"
		username := "root"
		if osType == "windows" {
			protocol = "rdp"
			username = "Administrator"
		}
		if osName == "" {
			osName = osType
		}
		status := "stopped"
		if inst.Status == "Running" {
			status = "running"
		}
		ipAddr := inst.PrivateIP
		if ipAddr == "" {
			ipAddr = inst.PublicIP
		}
		if ipAddr == "" {
			ipAddr = inst.InstanceID
		}
		hostname := inst.Hostname
		if hostname == "" {
			hostname = inst.InstanceName
		}
		if hostname == "" {
			hostname = inst.InstanceID
		}

		// 注意：servers 表的 cloud_id 外键引用的是 clouds 表(非 cloud_accounts)，
		// region_id 外键引用 regions 表。未找到关联时设为 NULL 避免外键约束失败。
		server := model.Server{
			Hostname:       hostname,
			IP:             ipAddr,
			PublicIP:       inst.PublicIP,
			Port:           defaultPort,
			Username:       username,
			AuthType:       "password",
			OSType:         osType,
			Protocol:       protocol,
			RDPPort:        3389,
			OS:             osName,
			CPU:            inst.CPU,
			Memory:         int64(inst.Memory),
			Status:         status,
			RegionID:       regionID, // nil = NULL
			CloudAccountID: &account.ID,
			Description:    "从 " + account.Provider + " 导入 (" + inst.InstanceID + ")",
			AgentStatus:    "offline",
		}
		if protocol == "rdp" && hasRDPCred {
			applyCredential(rdpCred, rdpPlain, &server)
		} else if protocol != "rdp" && hasSSHCred {
			applyCredential(sshCred, sshPlain, &server)
		}

		if err := h.db.Create(&server).Error; err == nil {
			imported++
		} else {
			skipped++
		}
	}

	c.JSON(http.StatusOK, model.Response{
		Code:    200,
		Message: "导入完成",
		Data: gin.H{
			"imported": imported,
			"skipped":  skipped,
		},
	})
}