diff --git a/backend/config/config.go b/backend/config/config.go index d59638a..168bff9 100644 --- a/backend/config/config.go +++ b/backend/config/config.go @@ -36,6 +36,7 @@ type Config struct { ActivateDailyLimit int // 单序列号 24 小时内新设备激活上限（0 = 不限制） OnlineTTLMinutes int // 激活记录在线状态有效期（last_seen 超过即视为离线） DeactivateRequireToken bool // 解绑是否强制校验设备激活令牌 + SyncIntervalMinutes int // 节点双向同步间隔（分钟），宽松节奏默认 10 // CORS 配置 CORSOrigins []string @@ -63,6 +64,7 @@ func Load() *Config { ActivateDailyLimit: getEnvInt("ACTIVATE_DAILY_LIMIT", 5), OnlineTTLMinutes: getEnvInt("ONLINE_TTL_MINUTES", 11520), // 默认 8 天，适配周检节奏 DeactivateRequireToken: getEnvBool("DEACTIVATE_REQUIRE_TOKEN", true), + SyncIntervalMinutes: getEnvInt("SYNC_INTERVAL_MINUTES", 10), CORSOrigins: defaultCORSOrigins(), } diff --git a/backend/database/database.go b/backend/database/database.go index a5a24d4..beb0a97 100644 --- a/backend/database/database.go +++ b/backend/database/database.go @@ -71,7 +71,9 @@ func migrateDatabase(db *gorm.DB) (*gorm.DB, error) { &model.Plan{}, // 定价计划 // Logan.Li &model.Order{}, // 订单 // Logan.Li // 防重复购买 - &model.CustomerEntitlement{}, // 客户权益 // Logan.Li + &model.CustomerEntitlement{}, // 客户权益 // Logan.Limodel.CustomerEntitlement{}, // 客户权益 // Logan.Li + // 节点互联 + &model.PeerNode{}, // 双向同步对端节点 // Logan.Li ) if err != nil { return nil, fmt.Errorf("数据库迁移失败: %v", err) diff --git a/backend/internal/controller/sync/sync_controller.go b/backend/internal/controller/sync/sync_controller.go new file mode 100644 index 0000000..95bae84 --- /dev/null +++ b/backend/internal/controller/sync/sync_controller.go @@ -0,0 +1,163 @@ +//! 节点互联控制器：管理端点（JWT）+ 节点间入站端点（HMAC）。 // Logan.Li + +package sync + +import ( + "backend/config" + "backend/internal/dto" + "backend/internal/model" + "encoding/json" + "io" + "net/http" + "strconv" + + "github.com/gin-gonic/gin" + "gorm.io/gorm" + + syncsvc "backend/internal/service/sync" +) + +// Controller 节点互联控制器 // Logan.Li +type Controller struct { + svc *syncsvc.Service + db *gorm.DB +} + +// NewController 创建控制器实例 // Logan.Li +func NewController(db *gorm.DB, cfg *config.Config) *Controller { + return &Controller{svc: syncsvc.NewService(db, cfg), db: db} +} + +// Svc 暴露 service 供路由层启动定时循环 // Logan.Li +func (ctrl *Controller) Svc() *syncsvc.Service { return ctrl.svc } + +// ============ 管理端点（JWT 保护） ============ + +// ListPeers 节点列表 // Logan.Li +func (ctrl *Controller) ListPeers(c *gin.Context) { + peers, err := ctrl.svc.ListPeers() + if err != nil { + c.JSON(http.StatusOK, gin.H{"code": 1, "message": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"code": 0, "data": gin.H{ + "list": peers, "node_id": ctrl.svc.NodeID(), + }}) +} + +// SavePeer 创建/更新节点 // Logan.Li +func (ctrl *Controller) SavePeer(c *gin.Context) { + var req dto.PeerRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusOK, gin.H{"code": 1, "message": "参数错误: " + err.Error()}) + return + } + peer, err := ctrl.svc.SavePeer(req) + if err != nil { + c.JSON(http.StatusOK, gin.H{"code": 1, "message": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"code": 0, "message": "已保存，请在对端配置相同互联密钥后测试握手", "data": peer}) +} + +// DeletePeer 删除节点 // Logan.Li +func (ctrl *Controller) DeletePeer(c *gin.Context) { + id, _ := strconv.ParseUint(c.Param("id"), 10, 32) + if err := ctrl.svc.DeletePeer(uint(id)); err != nil { + c.JSON(http.StatusOK, gin.H{"code": 1, "message": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"code": 0, "message": "已删除"}) +} + +// TestHandshake 手动握手测试 // Logan.Li +func (ctrl *Controller) TestHandshake(c *gin.Context) { + id, _ := strconv.ParseUint(c.Param("id"), 10, 32) + peer, err := ctrl.svc.TestHandshake(uint(id)) + if err != nil { + c.JSON(http.StatusOK, gin.H{"code": 1, "message": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"code": 0, "message": "握手成功，互联已激活", "data": peer}) +} + +// SyncNow 手动触发一轮双向同步 // Logan.Li +func (ctrl *Controller) SyncNow(c *gin.Context) { + id, _ := strconv.ParseUint(c.Param("id"), 10, 32) + var peer model.PeerNode + if err := ctrl.db.First(&peer, id).Error; err != nil { + c.JSON(http.StatusOK, gin.H{"code": 1, "message": "节点不存在"}) + return + } + result, err := ctrl.svc.ExchangeOut(&peer) + if err != nil { + c.JSON(http.StatusOK, gin.H{"code": 1, "message": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"code": 0, "message": "同步完成: " + result}) +} + +// GenerateKey 生成互联密钥 // Logan.Li +func (ctrl *Controller) GenerateKey(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{"code": 0, "data": gin.H{"shared_key": syncsvc.GenerateSharedKey()}}) +} + +// ============ 节点间入站端点（HMAC 保护，供对端调用） ============ + +const maxSyncBody = 32 << 20 + +// handshakeIn POST /api/v1/sync/handshake // Logan.Li +func (ctrl *Controller) HandshakeIn(c *gin.Context) { + body, peer := ctrl.readAndVerify(c) + if peer == nil { + return // readAndVerify 已写响应 // Logan.Li + } + var req dto.SyncHandshakeRequest + if err := json.Unmarshal(body, &req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"code": 1, "message": "请求体非法"}) + return + } + resp, err := ctrl.svc.HandshakeIn(peer, req) + if err != nil { + c.JSON(http.StatusUnauthorized, gin.H{"code": 1, "message": err.Error()}) + return + } + c.JSON(http.StatusOK, resp) +} + +// exchangeIn POST /api/v1/sync/exchange // Logan.Li +func (ctrl *Controller) ExchangeIn(c *gin.Context) { + body, peer := ctrl.readAndVerify(c) + if peer == nil { + return + } + var req dto.SyncExchangeRequest + if err := json.Unmarshal(body, &req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"code": 1, "message": "请求体非法"}) + return + } + resp, err := ctrl.svc.ExchangeIn(peer, req) + if err != nil { + c.JSON(http.StatusOK, gin.H{"code": 1, "message": err.Error()}) + return + } + c.JSON(http.StatusOK, resp) +} + +// readAndVerify 读取请求体并按互联密钥验签定位对端；失败时已写响应 // Logan.Li +func (ctrl *Controller) readAndVerify(c *gin.Context) ([]byte, *model.PeerNode) { + body, err := io.ReadAll(io.LimitReader(c.Request.Body, maxSyncBody)) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"code": 1, "message": "读取请求体失败"}) + return nil, nil + } + nodeID := c.Request.Header.Get("X-Sync-Node") + peer, err := ctrl.svc.ResolvePeer(nodeID, func(p *model.PeerNode) error { + return syncsvc.VerifySignature(p.SharedKey, c.Request, body) + }) + if err != nil { + c.JSON(http.StatusUnauthorized, gin.H{"code": 1, "message": err.Error()}) + return nil, nil + } + return body, peer +} diff --git a/backend/internal/dto/sync.go b/backend/internal/dto/sync.go new file mode 100644 index 0000000..1fbf905 --- /dev/null +++ b/backend/internal/dto/sync.go @@ -0,0 +1,118 @@ +package dto + +import "time" + +// ============ 互联节点管理（管理后台） ============ + +// PeerRequest 创建/更新互联节点 // Logan.Li +type PeerRequest struct { + ID uint `json:"id"` + Name string `json:"name" binding:"required"` + BaseURL string `json:"base_url" binding:"required,url"` + SharedKey string `json:"shared_key" binding:"required,min=32"` +} + +// PeerResponse 节点响应（不回传密钥明文） // Logan.Li +type PeerResponse struct { + ID uint `json:"id"` + Name string `json:"name"` + BaseURL string `json:"base_url"` + Status string `json:"status"` + RemoteNode string `json:"remote_node"` + KeyTail string `json:"key_tail"` // 密钥尾 4 位，供核对 // Logan.Li + LastSyncAt *time.Time `json:"last_sync_at"` + LastResult string `json:"last_result"` + CreatedAt time.Time `json:"created_at"` +} + +// ============ 节点间交换契约（HMAC 认证） ============ + +// SyncProductPayload 产品载荷（业务键 code） // Logan.Li +type SyncProductPayload struct { + Code string `json:"code"` + Name string `json:"name"` + Version string `json:"version"` + Description string `json:"description"` + Features string `json:"features"` + Price float64 `json:"price"` + PriceModel string `json:"price_model"` + Status int `json:"status"` + UpdatedAt time.Time `json:"updated_at"` +} + +// SyncPlanPayload 计划载荷（业务键 产品code+plan_type+duration） // Logan.Li +type SyncPlanPayload struct { + ProductCode string `json:"product_code"` + Name string `json:"name"` + PlanType string `json:"plan_type"` + Price float64 `json:"price"` + Currency string `json:"currency"` + Duration int `json:"duration"` + MaxDevices int `json:"max_devices"` + Features string `json:"features"` + SortOrder int `json:"sort_order"` + Status int `json:"status"` + UpdatedAt time.Time `json:"updated_at"` +} + +// SyncCustomerPayload 客户载荷（业务键 email，空邮箱回退 name）。 +// 合规最小化：不同步 phone/address/notes，仅同步邮箱场景必需字段 // Logan.Li +type SyncCustomerPayload struct { + Name string `json:"name"` + Company string `json:"company"` + Email string `json:"email"` + Status int `json:"status"` + UpdatedAt time.Time `json:"updated_at"` +} + +// SyncLicensePayload 授权载荷（业务键 license_key） // Logan.Li +type SyncLicensePayload struct { + LicenseKey string `json:"license_key"` + ProductCode string `json:"product_code"` + CustomerEmail string `json:"customer_email"` + CustomerName string `json:"customer_name"` + Status string `json:"status"` + StartDate time.Time `json:"start_date"` + EndDate time.Time `json:"end_date"` + Duration int `json:"duration"` + MaxActivations int `json:"max_activations"` + CurrentActivations int `json:"current_activations"` + IsTrial bool `json:"is_trial"` + TrialDays int `json:"trial_days"` + UpdatedAt time.Time `json:"updated_at"` +} + +// SyncItems 单次交换的数据集（按依赖顺序应用：产品→计划→客户→授权） // Logan.Li +type SyncItems struct { + Products []SyncProductPayload `json:"products"` + Plans []SyncPlanPayload `json:"plans"` + Customers []SyncCustomerPayload `json:"customers"` + Licenses []SyncLicensePayload `json:"licenses"` +} + +// SyncExchangeRequest 交换请求：携带我方增量 + 我方游标 // Logan.Li +type SyncExchangeRequest struct { + Cursor time.Time `json:"cursor"` // 对端上次同步到的时点（我方记录） + Items SyncItems `json:"items"` +} + +// SyncExchangeResponse 交换响应：对端增量 + 应用结果 + 对端时钟 // Logan.Li +type SyncExchangeResponse struct { + Items SyncItems `json:"items"` + Applied map[string]int `json:"applied"` + ServerTime time.Time `json:"server_time"` + NodeInfo map[string]string `json:"node_info"` +} + +// SyncHandshakeRequest 握手请求（验证互联密钥并互认节点） // Logan.Li +type SyncHandshakeRequest struct { + NodeName string `json:"node_name"` +} + +// SyncHandshakeResponse 握手响应 // Logan.Li +type SyncHandshakeResponse struct { + NodeID string `json:"node_id"` + NodeName string `json:"node_name"` + Version string `json:"version"` + ServerTime time.Time `json:"server_time"` +} diff --git a/backend/internal/model/sync.go b/backend/internal/model/sync.go new file mode 100644 index 0000000..480f96d --- /dev/null +++ b/backend/internal/model/sync.go @@ -0,0 +1,29 @@ +package model + +import "time" + +// Peer 节点状态：pending 已配置未握手，active 握手成功可同步，disabled 手动停用 // Logan.Li +const ( + PeerStatusPending = "pending" + PeerStatusActive = "active" + PeerStatusDisabled = "disabled" +) + +// PeerNode 互联节点：双向授权同步的对端（地址 + 互联密钥 + 游标） // Logan.Li +type PeerNode struct { + ID uint `json:"id" gorm:"primaryKey"` + Name string `json:"name" gorm:"size:100;not null"` + BaseURL string `json:"base_url" gorm:"size:255;not null"` // 对端 https 地址（含 /api） + SharedKey string `json:"-" gorm:"size:128;not null"` // 互联密钥，双向一致，HMAC 签名用 + Status string `json:"status" gorm:"size:20;default:'pending'"` + RemoteNode string `json:"remote_node" gorm:"size:64"` // 对端节点 ID，握手后回填 + LastSyncAt *time.Time `json:"last_sync_at,omitempty"` // 本地对端同步游标（增量起点） + LastResult string `json:"last_result" gorm:"size:255"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// TableName 指定表名 // Logan.Li +func (PeerNode) TableName() string { + return "peer_nodes" +} diff --git a/backend/internal/service/sync/sync_apply.go b/backend/internal/service/sync/sync_apply.go new file mode 100644 index 0000000..51e31b6 --- /dev/null +++ b/backend/internal/service/sync/sync_apply.go @@ -0,0 +1,173 @@ +//! 同步载荷应用：按业务键 upsert，仅当载荷新于本地时更新（last-write-wins）。 // Logan.Li + +package sync + +import ( + "fmt" + "strings" + "time" + + "backend/internal/dto" + "backend/internal/model" +) + +// upsertProduct 按产品 code 定位 // Logan.Li +func (s *Service) upsertProduct(p dto.SyncProductPayload) bool { + if strings.TrimSpace(p.Code) == "" { + return false + } + var local model.Product + err := s.db.Where("code = ?", p.Code).First(&local).Error + if err != nil { + local = model.Product{ + Code: p.Code, Name: p.Name, Version: p.Version, Description: p.Description, + Features: p.Features, Price: p.Price, PriceModel: p.PriceModel, + Status: p.Status, CreatedAt: p.UpdatedAt, UpdatedAt: p.UpdatedAt, + } + return s.db.Create(&local).Error == nil + } + if !p.UpdatedAt.After(local.UpdatedAt) { + return false // 本地已是最新的，跳过 // Logan.Li + } + local.Name = p.Name + local.Version = p.Version + local.Description = p.Description + local.Features = p.Features + local.Price = p.Price + local.PriceModel = p.PriceModel + local.Status = p.Status + return s.db.Save(&local).Error == nil +} + +// upsertPlan 按 产品code+plan_type+duration 定位 // Logan.Li +func (s *Service) upsertPlan(pl dto.SyncPlanPayload) bool { + if pl.ProductCode == "" { + return false + } + var product model.Product + if err := s.db.Where("code = ?", pl.ProductCode).First(&product).Error; err != nil { + return false // 产品依赖未就绪，等下一轮 // Logan.Li + } + var local model.Plan + err := s.db.Where("product_id = ? AND plan_type = ? AND duration = ?", product.ID, pl.PlanType, pl.Duration).First(&local).Error + if err != nil { + local = model.Plan{ + ProductID: product.ID, Name: pl.Name, PlanType: pl.PlanType, + Price: pl.Price, Currency: pl.Currency, Duration: pl.Duration, + MaxDevices: pl.MaxDevices, Features: pl.Features, SortOrder: pl.SortOrder, + Status: pl.Status, CreatedAt: pl.UpdatedAt, UpdatedAt: pl.UpdatedAt, + } + return s.db.Create(&local).Error == nil + } + if !pl.UpdatedAt.After(local.UpdatedAt) { + return false + } + local.Name = pl.Name + local.Price = pl.Price + local.Currency = pl.Currency + local.MaxDevices = pl.MaxDevices + local.Features = pl.Features + local.SortOrder = pl.SortOrder + local.Status = pl.Status + return s.db.Save(&local).Error == nil +} + +// upsertCustomer 按 email 定位（空/占位邮箱回退 name），合规最小字段 // Logan.Li +func (s *Service) upsertCustomer(c dto.SyncCustomerPayload) bool { + email := strings.TrimSpace(c.Email) + name := strings.TrimSpace(c.Name) + var local model.Customer + var err error + if email != "" && !strings.HasSuffix(email, "@local.invalid") { + err = s.db.Where("email = ?", email).First(&local).Error + } else { + if name == "" { + return false + } + err = s.db.Where("name = ?", name).Order("id ASC").First(&local).Error + } + if err != nil { + if email == "" { + email = fmt.Sprintf("sync+%d@local.invalid", time.Now().UnixNano()) + } + local = model.Customer{ + Name: name, Company: c.Company, Email: email, + Status: c.Status, CreatedAt: c.UpdatedAt, UpdatedAt: c.UpdatedAt, + } + return s.db.Create(&local).Error == nil + } + if !c.UpdatedAt.After(local.UpdatedAt) { + return false + } + local.Name = name + local.Company = c.Company + if email != "" && !strings.HasSuffix(email, "@local.invalid") { + local.Email = email + } + local.Status = c.Status + return s.db.Save(&local).Error == nil +} + +// upsertLicense 按 license_key 定位，跨表解析产品与客户 // Logan.Li +func (s *Service) upsertLicense(l dto.SyncLicensePayload) bool { + if strings.TrimSpace(l.LicenseKey) == "" || l.ProductCode == "" { + return false + } + var product model.Product + if err := s.db.Where("code = ?", l.ProductCode).First(&product).Error; err != nil { + return false + } + customerID, err := s.resolveCustomer(l.CustomerEmail, l.CustomerName) + if err != nil { + return false + } + var local model.License + err = s.db.Where("license_key = ?", l.LicenseKey).First(&local).Error + if err != nil { + local = model.License{ + LicenseKey: l.LicenseKey, ProductID: product.ID, CustomerID: customerID, + Status: l.Status, StartDate: l.StartDate, EndDate: l.EndDate, + Duration: l.Duration, MaxActivations: l.MaxActivations, + CurrentActivations: l.CurrentActivations, + IsTrial: l.IsTrial, TrialDays: l.TrialDays, + CreatedAt: l.UpdatedAt, UpdatedAt: l.UpdatedAt, + } + return s.db.Create(&local).Error == nil + } + if !l.UpdatedAt.After(local.UpdatedAt) { + return false + } + local.ProductID = product.ID + local.CustomerID = customerID + local.Status = l.Status + local.StartDate = l.StartDate + local.EndDate = l.EndDate + local.Duration = l.Duration + local.MaxActivations = l.MaxActivations + local.CurrentActivations = l.CurrentActivations + local.IsTrial = l.IsTrial + local.TrialDays = l.TrialDays + return s.db.Save(&local).Error == nil +} + +// resolveCustomer 按邮箱解析或创建客户（离线占位客户按名字复用） // Logan.Li +func (s *Service) resolveCustomer(email, name string) (uint, error) { + email = strings.TrimSpace(email) + name = strings.TrimSpace(name) + if email == "" { + email = fmt.Sprintf("sync+%d@local.invalid", time.Now().UnixNano()) + } + var customer model.Customer + if err := s.db.Where("email = ?", email).First(&customer).Error; err == nil { + return customer.ID, nil + } + customer = model.Customer{Name: name, Email: email, Status: 1} + if err := s.db.Create(&customer).Error; err != nil { + // 并发下可能已存在，再查一次 // Logan.Li + if err2 := s.db.Where("email = ?", email).First(&customer).Error; err2 == nil { + return customer.ID, nil + } + return 0, err + } + return customer.ID, nil +} diff --git a/backend/internal/service/sync/sync_exchange.go b/backend/internal/service/sync/sync_exchange.go new file mode 100644 index 0000000..1d015ed --- /dev/null +++ b/backend/internal/service/sync/sync_exchange.go @@ -0,0 +1,217 @@ +//! 同步交换：增量收集、按业务键 LWW 应用（产品→计划→客户→授权）、双向循环。 // Logan.Li + +package sync + +import ( + "encoding/json" + "fmt" + "log" + "time" + + "backend/internal/dto" + "backend/internal/model" +) + +// ResolvePeer 按签名定位对端：优先 remote_node 已回填记录，否则逐个验签（首次互联场景） // Logan.Li +func (s *Service) ResolvePeer(remoteNodeID string, verify func(*model.PeerNode) error) (*model.PeerNode, error) { + var peers []model.PeerNode + s.db.Where("remote_node = ?", remoteNodeID).Find(&peers) + if len(peers) == 0 { + s.db.Where("status != ?", model.PeerStatusDisabled).Find(&peers) + } + for i := range peers { + if err := verify(&peers[i]); err == nil { + // 首次验签成功即回填对端节点 ID // Logan.Li + if peers[i].RemoteNode == "" { + s.db.Model(&peers[i]).Update("remote_node", remoteNodeID) + } + return &peers[i], nil + } + } + return nil, fmt.Errorf("互联密钥校验失败（对端 node=%s）", remoteNodeID) +} + +// ============ 出站交换（主动方） ============ + +// ExchangeOut 与对端做一轮双向增量交换（手动按钮与定时循环共用） // Logan.Li +func (s *Service) ExchangeOut(peer *model.PeerNode) (string, error) { + s.mu.Lock() + defer s.mu.Unlock() + + cursor := time.Time{} + if peer.LastSyncAt != nil { + cursor = *peer.LastSyncAt + } + batchStart := time.Now() + items := s.gatherSince(cursor) + reqBody, _ := json.Marshal(dto.SyncExchangeRequest{Cursor: cursor, Items: items}) + + var resp dto.SyncExchangeResponse + if err := s.signedPost(peer, exchangePath, reqBody, &resp); err != nil { + result := "同步失败: " + truncate(err.Error(), 180) + s.db.Model(peer).Update("last_result", result) + return result, err + } + + applied := s.applyItems(resp.Items) + newCursor := batchStart + s.db.Model(peer).Updates(map[string]interface{}{ + "last_sync_at": &newCursor, + "last_result": fmt.Sprintf("双向同步成功 %s（对端时间 %s）", + formatApplied(applied), resp.ServerTime.Format("01-02 15:04")), + }) + return formatApplied(applied), nil +} + +// gatherSince 收集 updated_at 晚于游标的增量（依赖序：产品→计划→客户→授权） // Logan.Li +func (s *Service) gatherSince(cursor time.Time) dto.SyncItems { + var products []model.Product + s.db.Where("updated_at > ?", cursor).Order("updated_at ASC").Limit(syncBatchSize).Find(&products) + var plans []model.Plan + s.db.Where("updated_at > ?", cursor).Order("updated_at ASC").Limit(syncBatchSize).Find(&plans) + var customers []model.Customer + s.db.Where("updated_at > ?", cursor).Order("updated_at ASC").Limit(syncBatchSize).Find(&customers) + var licenses []model.License + s.db.Where("updated_at > ?", cursor).Order("updated_at ASC").Limit(syncBatchSize).Find(&licenses) + + items := dto.SyncItems{Products: []dto.SyncProductPayload{}, Plans: []dto.SyncPlanPayload{}, + Customers: []dto.SyncCustomerPayload{}, Licenses: []dto.SyncLicensePayload{}} + for _, p := range products { + items.Products = append(items.Products, dto.SyncProductPayload{ + Code: p.Code, Name: p.Name, Version: p.Version, Description: p.Description, + Features: p.Features, Price: p.Price, PriceModel: p.PriceModel, + Status: p.Status, UpdatedAt: p.UpdatedAt, + }) + } + productCodeByID := map[uint]string{} + for _, p := range products { + productCodeByID[p.ID] = p.Code + } + // 计划的产品可能不在本批增量里，补查全量映射 // Logan.Li + var allProducts []model.Product + s.db.Select("id, code").Find(&allProducts) + for _, p := range allProducts { + productCodeByID[p.ID] = p.Code + } + for _, pl := range plans { + items.Plans = append(items.Plans, dto.SyncPlanPayload{ + ProductCode: productCodeByID[pl.ProductID], Name: pl.Name, PlanType: pl.PlanType, + Price: pl.Price, Currency: pl.Currency, Duration: pl.Duration, + MaxDevices: pl.MaxDevices, Features: pl.Features, SortOrder: pl.SortOrder, + Status: pl.Status, UpdatedAt: pl.UpdatedAt, + }) + } + // 授权关联的客户信息补查（客户可能不在本批增量里） // Logan.Li + custByID := map[uint]model.Customer{} + var related []model.Customer + s.db.Find(&related) + for _, c := range related { + custByID[c.ID] = c + } + for _, l := range licenses { + c := custByID[l.CustomerID] + items.Licenses = append(items.Licenses, dto.SyncLicensePayload{ + LicenseKey: l.LicenseKey, ProductCode: productCodeByID[l.ProductID], + CustomerEmail: c.Email, CustomerName: c.Name, + Status: l.Status, StartDate: l.StartDate, EndDate: l.EndDate, + Duration: l.Duration, MaxActivations: l.MaxActivations, + CurrentActivations: l.CurrentActivations, + IsTrial: l.IsTrial, TrialDays: l.TrialDays, UpdatedAt: l.UpdatedAt, + }) + } + for _, c := range customers { + items.Customers = append(items.Customers, dto.SyncCustomerPayload{ + Name: c.Name, Company: c.Company, Email: c.Email, Status: c.Status, UpdatedAt: c.UpdatedAt, + }) + } + return items +} + +// ============ 入站应用（响应方） ============ + +// ExchangeIn 处理对端推送：应用其增量，返回我方增量 // Logan.Li +func (s *Service) ExchangeIn(peer *model.PeerNode, req dto.SyncExchangeRequest) (*dto.SyncExchangeResponse, error) { + applied := s.applyItems(req.Items) + + gatherStart := time.Now() + items := s.gatherSince(peerOutCursor(peer)) + s.db.Model(peer).Updates(map[string]interface{}{ + "last_sync_at": &gatherStart, + "last_result": "入站同步成功 " + formatApplied(applied), + }) + return &dto.SyncExchangeResponse{ + Items: items, Applied: applied, ServerTime: time.Now().UTC(), + NodeInfo: map[string]string{"node_id": s.nodeID, "name": s.hostName()}, + }, nil +} + +// peerOutCursor 出站游标即该 peer 记录的 last_sync_at（双向各自维护） // Logan.Li +func peerOutCursor(peer *model.PeerNode) time.Time { + if peer.LastSyncAt != nil { + return *peer.LastSyncAt + } + return time.Time{} +} + +// applyItems 按依赖顺序 LWW 应用（业务键定位，更新只在新于本地时生效） // Logan.Li +func (s *Service) applyItems(items dto.SyncItems) map[string]int { + applied := map[string]int{"products": 0, "plans": 0, "customers": 0, "licenses": 0, "skipped": 0} + for _, p := range items.Products { + if s.upsertProduct(p) { + applied["products"]++ + } else { + applied["skipped"]++ + } + } + for _, pl := range items.Plans { + if s.upsertPlan(pl) { + applied["plans"]++ + } else { + applied["skipped"]++ + } + } + for _, c := range items.Customers { + if s.upsertCustomer(c) { + applied["customers"]++ + } else { + applied["skipped"]++ + } + } + for _, l := range items.Licenses { + if s.upsertLicense(l) { + applied["licenses"]++ + } else { + applied["skipped"]++ + } + } + return applied +} + +func formatApplied(m map[string]int) string { + return fmt.Sprintf("产品%d/计划%d/客户%d/授权%d(跳过%d)", + m["products"], m["plans"], m["customers"], m["licenses"], m["skipped"]) +} + +// ============ 定时循环 ============ + +// StartSyncLoop 启动宽松节奏的定时双向同步（默认 10 分钟一轮，失败顺延下轮重试） // Logan.Li +func (s *Service) StartSyncLoop(intervalMinutes int) { + if intervalMinutes <= 0 { + intervalMinutes = 10 + } + go func() { + ticker := time.NewTicker(time.Duration(intervalMinutes) * time.Minute) + defer ticker.Stop() + for range ticker.C { + var peers []model.PeerNode + s.db.Where("status = ?", model.PeerStatusActive).Find(&peers) + for _, peer := range peers { + p := peer + if _, err := s.ExchangeOut(&p); err != nil { + log.Printf("[同步] 节点 %s 本轮失败: %v", peer.Name, err) + } + } + } + }() + log.Printf("[同步] 定时双向同步已启动（每 %d 分钟）", intervalMinutes) +} diff --git a/backend/internal/service/sync/sync_service.go b/backend/internal/service/sync/sync_service.go new file mode 100644 index 0000000..770dcab --- /dev/null +++ b/backend/internal/service/sync/sync_service.go @@ -0,0 +1,271 @@ +//! 节点互联服务：双向授权同步的 peer 管理、HMAC 认证、握手与增量交换。 +//! 加密通道 = 站点既有 HTTPS + 互联密钥 HMAC-SHA256 签名（防篡改/防重放），无需额外隧道。 // Logan.Li + +package sync + +import ( + "bytes" + "crypto/hmac" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "sync" + "time" + + "backend/config" + "backend/internal/dto" + "backend/internal/model" + + "gorm.io/gorm" +) + +const ( + syncBatchSize = 500 // 单实体单轮增量上限，宽松分批 // Logan.Li + handshakePath = "/v1/sync/handshake" + exchangePath = "/v1/sync/exchange" + sigTimeWindow = 5 * time.Minute // 签名时间戳容差（防重放） // Logan.Li + httpTimeout = 20 * time.Second + protocolVersion = "1" +) + +// Service 节点互联业务逻辑层 // Logan.Li +type Service struct { + db *gorm.DB + cfg *config.Config + nodeID string + client *http.Client + mu sync.Mutex // 同一时刻仅一轮交换，防游标错乱 // Logan.Li +} + +// NewService 创建 service 实例；节点 ID 由 JWT_SECRET 派生（两节点密钥不同，天然互异且稳定） // Logan.Li +func NewService(db *gorm.DB, cfg *config.Config) *Service { + sum := sha256.Sum256([]byte("opsx-sync-node:" + cfg.JWTSecret)) + return &Service{ + db: db, + cfg: cfg, + nodeID: hex.EncodeToString(sum[:8]), + client: &http.Client{Timeout: httpTimeout}, + } +} + +// NodeID 本节点标识 // Logan.Li +func (s *Service) NodeID() string { return s.nodeID } + +// GenerateSharedKey 生成 32 字节 hex 互联密钥 // Logan.Li +func GenerateSharedKey() string { + buf := make([]byte, 32) + if _, err := rand.Read(buf); err != nil { + panic("生成互联密钥失败: " + err.Error()) + } + return hex.EncodeToString(buf) +} + +// ============ Peer 管理 ============ + +// ListPeers 节点列表 // Logan.Li +func (s *Service) ListPeers() ([]dto.PeerResponse, error) { + var peers []model.PeerNode + if err := s.db.Order("id ASC").Find(&peers).Error; err != nil { + return nil, err + } + results := make([]dto.PeerResponse, 0, len(peers)) + for _, p := range peers { + results = append(results, toPeerResponse(p)) + } + return results, nil +} + +// SavePeer 创建或更新节点（ID=0 创建） // Logan.Li +func (s *Service) SavePeer(req dto.PeerRequest) (*dto.PeerResponse, error) { + base := normalizeBaseURL(req.BaseURL) + key := strings.TrimSpace(req.SharedKey) + if len(key) < 32 { + return nil, errors.New("互联密钥至少 32 位（建议用生成按钮）") + } + var peer model.PeerNode + if req.ID > 0 { + if err := s.db.First(&peer, req.ID).Error; err != nil { + return nil, errors.New("节点不存在") + } + peer.Name = req.Name + peer.BaseURL = base + peer.SharedKey = key + peer.Status = model.PeerStatusPending // 配置变更后需重新握手确认 // Logan.Li + } else { + var count int64 + s.db.Model(&model.PeerNode{}).Where("base_url = ?", base).Count(&count) + if count > 0 { + return nil, errors.New("该地址的节点已存在") + } + peer = model.PeerNode{Name: req.Name, BaseURL: base, SharedKey: key, Status: model.PeerStatusPending} + } + if err := s.db.Save(&peer).Error; err != nil { + return nil, err + } + resp := toPeerResponse(peer) + return &resp, nil +} + +// DeletePeer 删除节点 // Logan.Li +func (s *Service) DeletePeer(id uint) error { + return s.db.Delete(&model.PeerNode{}, id).Error +} + +func toPeerResponse(p model.PeerNode) dto.PeerResponse { + tail := "" + if len(p.SharedKey) >= 4 { + tail = p.SharedKey[len(p.SharedKey)-4:] + } + return dto.PeerResponse{ + ID: p.ID, Name: p.Name, BaseURL: p.BaseURL, Status: p.Status, + RemoteNode: p.RemoteNode, KeyTail: tail, + LastSyncAt: p.LastSyncAt, LastResult: p.LastResult, CreatedAt: p.CreatedAt, + } +} + +// normalizeBaseURL 统一为 https://host/api 形式；本地回环地址保留 http 供联调 // Logan.Li +func normalizeBaseURL(raw string) string { + base := strings.TrimRight(strings.TrimSpace(raw), "/") + if base == "" { + return base + } + isLocal := strings.Contains(base, "//127.0.0.1") || strings.Contains(base, "//localhost") + if strings.HasPrefix(base, "http://") && !isLocal { + base = "https://" + strings.TrimPrefix(base, "http://") + } + if strings.HasSuffix(base, "/api/v1") { + base = strings.TrimSuffix(base, "/v1") + } else if !strings.HasSuffix(base, "/api") { + base += "/api" + } + return base +} + +// ============ HMAC 签名 ============ + +// buildSignature 计算请求签名：HMAC-SHA256(key, "v1|node|ts|bodySha256") // Logan.Li +func buildSignature(key, nodeID string, ts int64, body []byte) string { + bodySum := sha256.Sum256(body) + mac := hmac.New(sha256.New, []byte(key)) + fmt.Fprintf(mac, "v1|%s|%d|%s", nodeID, ts, hex.EncodeToString(bodySum[:])) + return hex.EncodeToString(mac.Sum(nil)) +} + +// VerifySignature 校验入站请求签名与时间窗 // Logan.Li +func VerifySignature(key string, r *http.Request, body []byte) error { + nodeID := r.Header.Get("X-Sync-Node") + tsStr := r.Header.Get("X-Sync-Ts") + sig := r.Header.Get("X-Sync-Sig") + if nodeID == "" || tsStr == "" || sig == "" { + return errors.New("缺少同步签名头") + } + var ts int64 + if _, err := fmt.Sscanf(tsStr, "%d", &ts); err != nil { + return errors.New("签名时间戳非法") + } + if absDuration(time.Since(time.Unix(ts, 0))) > sigTimeWindow { + return errors.New("签名已过期（时钟偏差过大）") + } + expect := buildSignature(key, nodeID, ts, body) + if !hmac.Equal([]byte(expect), []byte(sig)) { + return errors.New("互联密钥校验失败") + } + return nil +} + +func absDuration(d time.Duration) time.Duration { + if d < 0 { + return -d + } + return d +} + +// ============ 握手 ============ + +// handshakeOut 对端发起握手 // Logan.Li +func (s *Service) handshakeOut(peer *model.PeerNode) (*dto.SyncHandshakeResponse, error) { + body, _ := json.Marshal(dto.SyncHandshakeRequest{NodeName: s.hostName()}) + var resp dto.SyncHandshakeResponse + if err := s.signedPost(peer, handshakePath, body, &resp); err != nil { + return nil, err + } + return &resp, nil +} + +// HandshakeIn 入站握手：签名已由 resolvePeer 按互联密钥验证，此处互认并激活 // Logan.Li +func (s *Service) HandshakeIn(peer *model.PeerNode, req dto.SyncHandshakeRequest) (*dto.SyncHandshakeResponse, error) { + if peer.Status == model.PeerStatusDisabled { + return nil, errors.New("本侧已停用该互联") + } + if peer.Status != model.PeerStatusActive { + s.db.Model(peer).Updates(map[string]interface{}{"status": model.PeerStatusActive}) + } + return &dto.SyncHandshakeResponse{ + NodeID: s.nodeID, NodeName: s.hostName(), Version: protocolVersion, ServerTime: time.Now().UTC(), + }, nil +} + +// TestHandshake 管理端手动握手：成功则双方激活 // Logan.Li +func (s *Service) TestHandshake(id uint) (*dto.PeerResponse, error) { + var peer model.PeerNode + if err := s.db.First(&peer, id).Error; err != nil { + return nil, errors.New("节点不存在") + } + resp, err := s.handshakeOut(&peer) + if err != nil { + s.db.Model(&peer).Updates(map[string]interface{}{"last_result": "握手失败: " + truncate(err.Error(), 200)}) + return nil, fmt.Errorf("握手失败: %s", err.Error()) + } + s.db.Model(&peer).Updates(map[string]interface{}{ + "status": model.PeerStatusActive, "remote_node": resp.NodeID, + "last_result": "握手成功，对端: " + resp.NodeName, + }) + _ = s.db.First(&peer, id).Error + out := toPeerResponse(peer) + return &out, nil +} + +// signedPost 携带 HMAC 签名调用对端 // Logan.Li +func (s *Service) signedPost(peer *model.PeerNode, path string, body []byte, out interface{}) error { + url := strings.TrimRight(peer.BaseURL, "/") + path + ts := time.Now().Unix() + req, err := http.NewRequest("POST", url, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Sync-Node", s.nodeID) + req.Header.Set("X-Sync-Ts", fmt.Sprintf("%d", ts)) + req.Header.Set("X-Sync-Sig", buildSignature(peer.SharedKey, s.nodeID, ts, body)) + resp, err := s.client.Do(req) + if err != nil { + return fmt.Errorf("请求对端失败: %w", err) + } + defer resp.Body.Close() + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 32<<20)) + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("对端返回 %d: %s", resp.StatusCode, truncate(string(raw), 120)) + } + if err := json.Unmarshal(raw, out); err != nil { + return fmt.Errorf("对端响应解析失败: %w", err) + } + return nil +} + +// hostName 对端展示用本机名（无 os/env 依赖，用节点 ID 前缀） // Logan.Li +func (s *Service) hostName() string { + return "node-" + s.nodeID[:8] +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] +} diff --git a/backend/router/router.go b/backend/router/router.go index b010daa..397815c 100644 --- a/backend/router/router.go +++ b/backend/router/router.go @@ -11,6 +11,7 @@ import ( "backend/internal/controller/plan" "backend/internal/controller/product" "backend/internal/controller/setting" + synccontroller "backend/internal/controller/sync" "backend/middleware" "strings" @@ -93,9 +94,20 @@ func SetupRouter(db *gorm.DB, cfg *config.Config) *gin.Engine { entitlementController := entitlement.NewEntitlementController(db) analyticsController := analytics.NewAnalyticsController(db) settingController := setting.NewSettingController(db) + syncController := synccontroller.NewController(db, cfg) + + // 启动节点定时双向同步（宽松节奏，默认 10 分钟） // Logan.Li + syncController.Svc().StartSyncLoop(cfg.SyncIntervalMinutes) api := r.Group("/api") { + // ============ 节点互联（对端调用，HMAC 签名认证，非 JWT） ============ + syncGroup := api.Group("/v1/sync") + { + syncGroup.POST("/handshake", middleware.RateLimit(1, 30), syncController.HandshakeIn) + syncGroup.POST("/exchange", middleware.RateLimit(2, 60), syncController.ExchangeIn) + } + // ============ 认证路由（无需 JWT） ============ authGroup := api.Group("/auth") { @@ -249,6 +261,17 @@ func SetupRouter(db *gorm.DB, cfg *config.Config) *gin.Engine { register(system, "POST", "/waffo/test", settingController.DebugWaffo) } + // 节点互联管理（双向授权同步） // Logan.Li + syncAdmin := protected.Group("/sync") + { + register(syncAdmin, "GET", "/peers", syncController.ListPeers) + register(syncAdmin, "POST", "/peers", syncController.SavePeer) + register(syncAdmin, "DELETE", "/peers/:id", syncController.DeletePeer) + register(syncAdmin, "POST", "/peers/:id/handshake", syncController.TestHandshake) + register(syncAdmin, "POST", "/peers/:id/sync", syncController.SyncNow) + register(syncAdmin, "GET", "/generate-key", syncController.GenerateKey) + } + // 支付调试（与 settings/waffo/test 同一 handler） // Logan.Li paymentGroup := protected.Group("/payment") { diff --git a/frontend/src/api/sync.js b/frontend/src/api/sync.js new file mode 100644 index 0000000..c1ae7b9 --- /dev/null +++ b/frontend/src/api/sync.js @@ -0,0 +1,29 @@ +/** + * @file sync.js + * @description 节点互联（双向授权同步）管理 API // Logan.Li + */ +import request from './request' + +export function listPeers() { + return request.get('/sync/peers') +} + +export function savePeer(data) { + return request.post('/sync/peers', data) +} + +export function deletePeer(id) { + return request.delete(`/sync/peers/${id}`) +} + +export function handshakePeer(id) { + return request.post(`/sync/peers/${id}/handshake`) +} + +export function syncPeerNow(id) { + return request.post(`/sync/peers/${id}/sync`) +} + +export function generateKey() { + return request.get('/sync/generate-key') +} diff --git a/frontend/src/layouts/MainLayout.vue b/frontend/src/layouts/MainLayout.vue index f868393..194f3b0 100644 --- a/frontend/src/layouts/MainLayout.vue +++ b/frontend/src/layouts/MainLayout.vue @@ -56,6 +56,11 @@ + + + + +