feat(server): add PCDN scheduling & dispatch pipeline

server/initialize/gorm_biz.go

@@ -5,12 +5,13 @@ import (
 	"github.com/flipped-aurora/gin-vue-admin/server/model/computenode"
 	"github.com/flipped-aurora/gin-vue-admin/server/model/imageregistry"
 	"github.com/flipped-aurora/gin-vue-admin/server/model/instance"
+	"github.com/flipped-aurora/gin-vue-admin/server/model/pcdn"
 	"github.com/flipped-aurora/gin-vue-admin/server/model/product"
 )
 
 func bizModel() error {
 	db := global.GVA_DB
-	err := db.AutoMigrate(imageregistry.ImageRegistry{}, computenode.ComputeNode{}, product.ProductSpec{}, instance.Instance{})
+	err := db.AutoMigrate(imageregistry.ImageRegistry{}, computenode.ComputeNode{}, product.ProductSpec{}, instance.Instance{}, pcdn.PcdnDispatchTask{})
 	if err != nil {
 		return err
 	}

server/initialize/timer.go

@@ -32,6 +32,20 @@ func Timer() {
 
 		// 其他定时任务定在这里 参考上方使用方法
 
+		_, err = global.GVA_Timer.AddTaskByFuncWithSecond("PCDNDispatch", "*/10 * * * * *", func() {
+			task.ProcessPcdnDispatchTasks()
+		}, "PCDN调度任务执行", option...)
+		if err != nil {
+			fmt.Println("add PCDN dispatch timer error:", err)
+		}
+
+		_, err = global.GVA_Timer.AddTaskByFuncWithSecond("PCDNDispatch", "*/12 * * * * *", func() {
+			task.SyncPcdnDispatchStatus()
+		}, "PCDN调度状态同步", option...)
+		if err != nil {
+			fmt.Println("add PCDN sync timer error:", err)
+		}
+
 		//_, err := global.GVA_Timer.AddTaskByFunc("定时任务标识", "corn表达式", func() {
 		//	具体执行内容...
 		//  ......

server/model/pcdn/dispatch_task.go

@@ -0,0 +1,40 @@
+package pcdn
+
+import (
+	"github.com/flipped-aurora/gin-vue-admin/server/global"
+	"github.com/flipped-aurora/gin-vue-admin/server/model/common"
+)
+
+const (
+	DispatchStatusPending  = "pending"
+	DispatchStatusRunning  = "running"
+	DispatchStatusSuccess  = "success"
+	DispatchStatusFailed   = "failed"
+	DispatchStatusRetrying = "retrying"
+)
+
+// PcdnDispatchTask 记录调度计划及下发执行结果，支持审计与重放。
+type PcdnDispatchTask struct {
+	global.GVA_MODEL
+	TaskID           string         `json:"taskId" gorm:"column:task_id;size:64;not null;uniqueIndex:uk_task_trace,priority:1;index:idx_task_trace,priority:1"`
+	TraceID          string         `json:"traceId" gorm:"column:trace_id;size:64;not null;uniqueIndex:uk_task_trace,priority:2;index:idx_task_trace,priority:2"`
+	ContentID        string         `json:"contentId" gorm:"column:content_id;size:128;not null;index"`
+	UserRegion       string         `json:"userRegion" gorm:"column:user_region;size:64;not null;index"`
+	UserISP          string         `json:"userIsp" gorm:"column:user_isp;size:64;default:''"`
+	TopN             int            `json:"topN" gorm:"column:top_n;default:3"`
+	Status           string         `json:"status" gorm:"column:status;size:32;not null;index"`
+	RetryCount       int            `json:"retryCount" gorm:"column:retry_count;default:0"`
+	MaxRetry         int            `json:"maxRetry" gorm:"column:max_retry;default:3"`
+	TimeoutSeconds   int            `json:"timeoutSeconds" gorm:"column:timeout_seconds;default:8"`
+	NextRetryUnix    int64          `json:"nextRetryUnix" gorm:"column:next_retry_unix;default:0;index"`
+	LastError        string         `json:"lastError" gorm:"column:last_error;size:1024;default:''"`
+	Candidates       common.JSONMap `json:"candidates" gorm:"column:candidates;type:json"`
+	MetricsSnapshot  common.JSONMap `json:"metricsSnapshot" gorm:"column:metrics_snapshot;type:json"`
+	DispatchProtocol string         `json:"dispatchProtocol" gorm:"column:dispatch_protocol;size:64;default:'mock'"`
+	PrimaryNodeID    uint           `json:"primaryNodeId" gorm:"column:primary_node_id;default:0"`
+	CurrentNodeID    uint           `json:"currentNodeId" gorm:"column:current_node_id;default:0"`
+}
+
+func (PcdnDispatchTask) TableName() string {
+	return "pcdn_dispatch_task"
+}

server/service/enter.go

@@ -5,6 +5,7 @@ import (
 	"github.com/flipped-aurora/gin-vue-admin/server/service/example"
 	"github.com/flipped-aurora/gin-vue-admin/server/service/imageregistry"
 	"github.com/flipped-aurora/gin-vue-admin/server/service/instance"
+	"github.com/flipped-aurora/gin-vue-admin/server/service/pcdn"
 	"github.com/flipped-aurora/gin-vue-admin/server/service/product"
 	"github.com/flipped-aurora/gin-vue-admin/server/service/system"
 )
@@ -18,4 +19,5 @@ type ServiceGroup struct {
 	ComputenodeServiceGroup   computenode.ServiceGroup
 	ProductServiceGroup       product.ServiceGroup
 	InstanceServiceGroup      instance.ServiceGroup
+	PcdnServiceGroup          pcdn.ServiceGroup
 }

server/service/pcdn/dispatcher.go

@@ -0,0 +1,50 @@
+package pcdn
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// DispatchRequest 下发请求。
+type DispatchRequest struct {
+	TaskID     string
+	TraceID    string
+	ContentID  string
+	TargetNode uint
+	TimeoutSec int
+}
+
+// DispatchResult 下发结果。
+type DispatchResult struct {
+	Success bool
+	Message string
+}
+
+// Dispatcher 抽象下发协议接口，便于后续扩展 HTTP/gRPC/MQ。
+type Dispatcher interface {
+	Dispatch(ctx context.Context, request DispatchRequest) DispatchResult
+	Protocol() string
+}
+
+// MockDispatcher 为默认实现。
+type MockDispatcher struct{}
+
+func (d *MockDispatcher) Protocol() string { return "mock" }
+
+func (d *MockDispatcher) Dispatch(ctx context.Context, request DispatchRequest) DispatchResult {
+	if request.TimeoutSec <= 0 {
+		request.TimeoutSec = 8
+	}
+	timeoutCtx, cancel := context.WithTimeout(ctx, time.Duration(request.TimeoutSec)*time.Second)
+	defer cancel()
+	select {
+	case <-timeoutCtx.Done():
+		return DispatchResult{Success: false, Message: timeoutCtx.Err().Error()}
+	case <-time.After(50 * time.Millisecond):
+		if request.TargetNode == 0 {
+			return DispatchResult{Success: false, Message: "invalid target node"}
+		}
+		return DispatchResult{Success: true, Message: fmt.Sprintf("dispatched to node %d", request.TargetNode)}
+	}
+}

server/service/pcdn/enter.go

@@ -0,0 +1,5 @@
+package pcdn
+
+type ServiceGroup struct {
+	SchedulerService SchedulerService
+}

server/service/pcdn/fallback.go

@@ -0,0 +1,56 @@
+package pcdn
+
+import pcdnModel "github.com/flipped-aurora/gin-vue-admin/server/model/pcdn"
+
+// NextFallbackNode 主节点失败时选择次优候选。
+func NextFallbackNode(task pcdnModel.PcdnDispatchTask) uint {
+	raw, ok := task.Candidates["list"]
+	if ok {
+		if nodeID := findNextNode(raw, task.CurrentNodeID); nodeID != 0 {
+			return nodeID
+		}
+	}
+	return findNextNode(task.Candidates, task.CurrentNodeID)
+}
+
+func findNextNode(raw any, current uint) uint {
+	switch list := raw.(type) {
+	case []any:
+		for _, item := range list {
+			nodeID := parseNodeID(item)
+			if nodeID != 0 && nodeID != current {
+				return nodeID
+			}
+		}
+	case map[string]any:
+		nodeID := parseNodeID(list)
+		if nodeID != 0 && nodeID != current {
+			return nodeID
+		}
+	}
+	return 0
+}
+
+func parseNodeID(v any) uint {
+	m, ok := v.(map[string]any)
+	if !ok {
+		return 0
+	}
+	raw, ok := m["node_id"]
+	if !ok {
+		return 0
+	}
+	switch id := raw.(type) {
+	case uint:
+		return id
+	case int:
+		if id > 0 {
+			return uint(id)
+		}
+	case float64:
+		if id > 0 {
+			return uint(id)
+		}
+	}
+	return 0
+}

server/service/pcdn/policy_engine.go

@@ -0,0 +1,48 @@
+package pcdn
+
+import "math"
+
+// RealtimeNodeMetric 节点实时指标。
+type RealtimeNodeMetric struct {
+	NodeID         uint
+	LatencyMS      float64
+	UnitCost       float64
+	LoadPercent    float64
+	HealthScore    float64
+	Online         bool
+	PolicyDisabled bool
+	ISP            string
+}
+
+// ScoreWeights 策略权重。
+type ScoreWeights struct {
+	Latency float64
+	Cost    float64
+	Load    float64
+	Health  float64
+}
+
+// PolicyEngine 执行加权评分。
+type PolicyEngine struct {
+	weights ScoreWeights
+}
+
+func NewPolicyEngine(weights ScoreWeights) *PolicyEngine {
+	if weights == (ScoreWeights{}) {
+		weights = ScoreWeights{Latency: 0.35, Cost: 0.2, Load: 0.2, Health: 0.25}
+	}
+	return &PolicyEngine{weights: weights}
+}
+
+// Score 将指标归一化后进行加权，分数越高越优。
+func (p *PolicyEngine) Score(metric RealtimeNodeMetric) float64 {
+	latencyScore := 1 / (1 + math.Max(metric.LatencyMS, 0)/50)
+	costScore := 1 / (1 + math.Max(metric.UnitCost, 0))
+	loadScore := 1 - math.Min(math.Max(metric.LoadPercent, 0), 100)/100
+	healthScore := math.Min(math.Max(metric.HealthScore, 0), 100) / 100
+
+	return p.weights.Latency*latencyScore +
+		p.weights.Cost*costScore +
+		p.weights.Load*loadScore +
+		p.weights.Health*healthScore
+}