244 lines
8.3 KiB
Go
244 lines
8.3 KiB
Go
//! 同步交换：增量收集、按业务键 LWW 应用（产品→计划→客户→授权）、双向循环。 // Logan.Li
|
||
|
||
package sync
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"log"
|
||
"sync/atomic"
|
||
"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 分钟一轮）。
|
||
// 行为：启动即先跑一轮；active 节点直接交换；pending 节点先自动握手（两侧
|
||
// 均已配置同钥时自动激活，无需人工点击）；上一轮未结束时本轮跳过（防堆积）。 // Logan.Li
|
||
func (s *Service) StartSyncLoop(intervalMinutes int) {
|
||
if intervalMinutes <= 0 {
|
||
intervalMinutes = 10
|
||
}
|
||
go func() {
|
||
var running int32
|
||
runOnce := func() {
|
||
if !atomic.CompareAndSwapInt32(&running, 0, 1) {
|
||
return // 上一轮仍在进行，跳过本轮防堆积 // Logan.Li
|
||
}
|
||
defer atomic.StoreInt32(&running, 0)
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
log.Printf("[同步] 本轮异常恢复: %v", r)
|
||
}
|
||
}()
|
||
|
||
var peers []model.PeerNode
|
||
s.db.Where("status IN ?", []string{model.PeerStatusActive, model.PeerStatusPending}).Find(&peers)
|
||
for _, peer := range peers {
|
||
p := peer
|
||
// pending 先尝试握手（对端未配置/密钥不一致时静默失败，下轮再试） // Logan.Li
|
||
if p.Status == model.PeerStatusPending {
|
||
if _, err := s.TestHandshake(p.ID); err != nil {
|
||
continue
|
||
}
|
||
s.db.First(&p, p.ID)
|
||
}
|
||
if _, err := s.ExchangeOut(&p); err != nil {
|
||
log.Printf("[同步] 节点 %s 本轮失败: %v", peer.Name, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
runOnce() // 启动即先跑一轮，不等首个间隔 // Logan.Li
|
||
ticker := time.NewTicker(time.Duration(intervalMinutes) * time.Minute)
|
||
defer ticker.Stop()
|
||
for range ticker.C {
|
||
runOnce()
|
||
}
|
||
}()
|
||
log.Printf("[同步] 定时双向同步已启动（每 %d 分钟）", intervalMinutes)
|
||
}
|