feat: 节点互联双向同步——授权/计划/客户(邮箱)按业务键LWW增量同步，HTTPS+HMAC互联密钥，默认10分钟宽松节奏
This commit is contained in:
Logan.Li committed
1 parent
06a9dac99b
commit
44387835e1
13 files changed
+1180
-1
No files matched your search
@@ -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)
|
||||
}
|
||||
Reference in new issue
Block a user