opsx-vpn/backend/internal/api/batch.go

package api

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

	"github.com/gin-gonic/gin"

	"vpn-platform/backend/internal/model"
	"vpn-platform/backend/internal/service"
	"vpn-platform/backend/internal/store"
	"vpn-platform/backend/internal/taskmgr"
)

// batchRequest 批量操作请求体
type batchRequest struct {
	TunnelIDs []uint `json:"tunnel_ids" binding:"required,min=1"`
}

// batchTunnels 批量启动/停止/删除隧道
func (s *Server) batchTunnels(action taskmgr.TaskAction) gin.HandlerFunc {
	return func(c *gin.Context) {
		var req batchRequest
		if err := c.ShouldBindJSON(&req); err != nil {
			c.JSON(http.StatusBadRequest, gin.H{"error": "参数错误: " + err.Error()})
			return
		}

		// 验证所有隧道存在
		var tunnels []model.Tunnel
		if err := store.DB.Where("id IN ?", req.TunnelIDs).Find(&tunnels).Error; err != nil {
			c.JSON(http.StatusInternalServerError, gin.H{"error": "查询隧道失败"})
			return
		}
		if len(tunnels) != len(req.TunnelIDs) {
			found := map[uint]bool{}
			for _, t := range tunnels {
				found[t.ID] = true
			}
			var missing []uint
			for _, id := range req.TunnelIDs {
				if !found[id] {
					missing = append(missing, id)
				}
			}
			c.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("隧道不存在: %v", missing)})
			return
		}

		// 构建名称映射
		names := map[uint]string{}
		for _, t := range tunnels {
			names[t.ID] = t.Name
		}

		// 创建任务
		taskID := s.taskMgr.Create(action, req.TunnelIDs, names)
		username := c.GetString("username")

		// 并发执行
		go func() {
			var wg sync.WaitGroup
			sem := make(chan struct{}, 5) // 最多 5 个并发

			for _, tid := range req.TunnelIDs {
				wg.Add(1)
				sem <- struct{}{}
				go func(tunnelID uint) {
					defer wg.Done()
					defer func() { <-sem }()

					s.taskMgr.MarkRunning(taskID, tunnelID)
					err := s.executeBatchAction(action, tunnelID, username)
					if err != nil {
						s.taskMgr.MarkFailed(taskID, tunnelID, err)
					} else {
						s.taskMgr.MarkDone(taskID, tunnelID)
					}
				}(tid)
			}
			wg.Wait()
		}()

		c.JSON(http.StatusAccepted, gin.H{"task_id": taskID})
	}
}

// executeBatchAction 执行单个隧道的操作
func (s *Server) executeBatchAction(action taskmgr.TaskAction, tunnelID uint, username string) error {
	var t model.Tunnel
	if err := store.DB.First(&t, tunnelID).Error; err != nil {
		return fmt.Errorf("隧道不存在")
	}

	mu := lockTunnel(t.ID)
	defer mu.Unlock()

	switch action {
	case taskmgr.ActionStart:
		if t.ExpireAt != nil && time.Now().After(*t.ExpireAt) {
			return fmt.Errorf("专线已到期")
		}
		p, err := s.mgr.Get(t.Protocol)
		if err != nil {
			return err
		}
		if t.State == "running" {
			return nil // 已经运行中，视为成功
		}
		if err := p.Start(&t); err != nil {
			t.State = "error"
			t.LastError = err.Error()
			store.DB.Save(&t)
			return err
		}
		t.State = "running"
		t.LastError = ""
		store.DB.Save(&t)
		store.Audit(username, "batch_start", t.Name, "")
		if err := s.syncAuthGatewayForTunnel(&t); err != nil {
			return fmt.Errorf("专线已启动，但公网认证入口同步失败: %w", err)
		}

	case taskmgr.ActionStop:
		p, err := s.mgr.Get(t.Protocol)
		if err != nil {
			return err
		}
		if t.State == "stopped" {
			return nil
		}
		if err := p.Stop(&t); err != nil {
			return err
		}
		t.State = "stopped"
		store.DB.Save(&t)
		store.Audit(username, "batch_stop", t.Name, "")
		if err := s.syncAuthGatewayForTunnel(&t); err != nil {
			return fmt.Errorf("专线已停止，但公网认证入口同步失败: %w", err)
		}

	case taskmgr.ActionDelete:
		if p, err := s.mgr.Get(t.Protocol); err == nil {
			_ = p.Delete(&t)
		}
		if err := service.RemoveInstance(&t); err != nil && !os.IsNotExist(err) {
			return fmt.Errorf("清理实例文件失败: %w", err)
		}
		s.mgr.RemoveStatus(t.ID)
		store.DB.Delete(&t)
		store.Audit(username, "batch_delete", t.Name, "")
		if err := s.syncAuthGatewayForTunnel(&t); err != nil {
			return fmt.Errorf("专线已删除，但公网认证入口同步失败: %w", err)
		}
	}

	return nil
}

// getTask 查询单个任务状态
func (s *Server) getTask(c *gin.Context) {
	id := c.Param("id")
	t, ok := s.taskMgr.Get(id)
	if !ok {
		c.JSON(http.StatusNotFound, gin.H{"error": "任务不存在"})
		return
	}
	c.JSON(http.StatusOK, t)
}

// listTasks 列出最近任务
func (s *Server) listTasks(c *gin.Context) {
	tasks := s.taskMgr.List()
	c.JSON(http.StatusOK, tasks)
}

// ensureID 确保 ID 有效（从 batchCreateUsers 模式参考）
func parseTunnelIDs(c *gin.Context) ([]uint, error) {
	var req batchRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		return nil, err
	}
	for _, id := range req.TunnelIDs {
		if id == 0 {
			return nil, fmt.Errorf("无效的隧道 ID: %d", id)
		}
	}
	return req.TunnelIDs, nil
}

// batchTrafficPolicy 批量设置流量策略
type batchPolicyRequest struct {
	TunnelIDs      []uint `json:"tunnel_ids" binding:"required,min=1"`
	RateLimitMbps  int    `json:"rate_limit_mbps"`
	QuotaBytes     int64  `json:"quota_bytes"`
	ActionOnExceed string `json:"action_on_exceed"`
	ResetCycle     string `json:"reset_cycle"`
}

func (s *Server) batchSetPolicy(c *gin.Context) {
	var req batchPolicyRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "参数错误: " + err.Error()})
		return
	}

	success := 0
	var errors []string
	for _, tid := range req.TunnelIDs {
		var t model.Tunnel
		if err := store.DB.First(&t, tid).Error; err != nil {
			errors = append(errors, fmt.Sprintf("隧道 %d 不存在", tid))
			continue
		}
		var policy model.TrafficPolicy
		if err := store.DB.Where("tunnel_id = ?", tid).First(&policy).Error; err != nil {
			policy.TunnelID = tid
		}
		if req.RateLimitMbps > 0 {
			policy.RateLimitMbps = req.RateLimitMbps
		}
		if req.QuotaBytes > 0 {
			policy.QuotaBytes = req.QuotaBytes
		}
		if req.ActionOnExceed != "" {
			policy.ActionOnExceed = req.ActionOnExceed
		}
		if req.ResetCycle != "" {
			policy.ResetCycle = req.ResetCycle
		}
		if err := store.DB.Save(&policy).Error; err != nil {
			errors = append(errors, fmt.Sprintf("隧道 %s 策略保存失败: %s", t.Name, err.Error()))
			continue
		}
		success++
		store.Audit(c.GetString("username"), "batch_policy", t.Name, strconv.Itoa(req.RateLimitMbps)+"Mbps")
	}

	c.JSON(http.StatusOK, gin.H{
		"success": success,
		"errors":  errors,
	})
}