LinkPort/internal/store/sync_bundle.go

package store

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"sort"
)

// NormalizeSyncBundle removes device runtime/identity fields and gives the same
// configuration a stable hash on machines with different vault keys.
func NormalizeSyncBundle(b *Bundle) error {
	for i := range b.Gateways {
		b.Gateways[i].Enabled = false
		b.Gateways[i].SSH.HostKey = ""
	}
	for i := range b.Assets {
		b.Assets[i].UpdatedAt = ""
	}
	if err := validateBundle(b); err != nil {
		return errors.New("同步配置校验失败，请检查资源及引用")
	}
	sort.Slice(b.Assets, func(i, j int) bool { return b.Assets[i].ID < b.Assets[j].ID })
	sort.Slice(b.Targets, func(i, j int) bool { return b.Targets[i].ID < b.Targets[j].ID })
	sort.Slice(b.Outbounds, func(i, j int) bool { return b.Outbounds[i].ID < b.Outbounds[j].ID })
	sort.Slice(b.Gateways, func(i, j int) bool { return b.Gateways[i].ID < b.Gateways[j].ID })
	// Canonical empty arrays regardless of how a remote JSON encoder emits them.
	if len(b.Assets) == 0 {
		b.Assets = nil
	}
	if len(b.Targets) == 0 {
		b.Targets = nil
	}
	if len(b.Outbounds) == 0 {
		b.Outbounds = nil
	}
	if len(b.Gateways) == 0 {
		b.Gateways = nil
	}
	return nil
}

func SyncHash(b Bundle) string {
	raw, _ := json.Marshal(b)
	hash := sha256.Sum256(raw)
	return hex.EncodeToString(hash[:])
}

func (s *Store) SyncSnapshot() (Bundle, string, error) {
	tx, err := s.db.Begin()
	if err != nil {
		return Bundle{}, "", err
	}
	defer tx.Rollback()
	b, err := s.bundle(tx)
	if err != nil {
		return b, "", err
	}
	revision, err := configurationRevision(tx)
	if err != nil {
		return b, "", err
	}
	err = NormalizeSyncBundle(&b)
	return b, revision, err
}

// ReplaceSyncBundle propagates deletions as well as changes, and commits the
// baseline atomically with the imported configuration. Local history is retained.
func (s *Store) ReplaceSyncBundle(b Bundle, revision, hash string) error {
	if err := NormalizeSyncBundle(&b); err != nil {
		return err
	}
	tx, err := s.db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	current, err := configurationRevision(tx)
	if err != nil {
		return err
	}
	if current != revision {
		return ErrConfigurationChanged
	}
	if err := s.saveScopeCatalog(tx, b.Scopes); err != nil {
		return err
	}
	for _, kind := range []string{"gateways", "targets", "outbounds", "assets"} {
		rows, e := tx.Query("SELECT id FROM " + kind)
		if e != nil {
			return e
		}
		var ids []string
		for rows.Next() {
			var id string
			if e = rows.Scan(&id); e != nil {
				rows.Close()
				return e
			}
			ids = append(ids, id)
		}
		e = rows.Err()
		rows.Close()
		if e != nil {
			return e
		}
		for _, id := range ids {
			if err = s.recordHistory(tx, kind, id); err != nil {
				return err
			}
		}
		if _, err = tx.Exec("DELETE FROM " + kind); err != nil {
			return err
		}
	}
	for _, a := range b.Assets {
		if err = s.saveAssetFromBundle(tx, a); err != nil {
			return err
		}
	}
	for _, o := range b.Outbounds {
		if err = s.saveOutbound(tx, o); err != nil {
			return err
		}
	}
	for _, t := range b.Targets {
		if err = s.saveTarget(tx, t); err != nil {
			return err
		}
	}
	for _, g := range b.Gateways {
		if err = s.saveGateway(tx, g); err != nil {
			return err
		}
	}
	_, err = tx.Exec("UPDATE webdav_sync_state SET base_hash=?,conflict=0,local_hash='',remote_hash='',action='pulled' WHERE id=1", hash)
	if err != nil {
		return err
	}
	return tx.Commit()
}

func (s *Store) SetSyncBaseline(hash, action string) error {
	_, err := s.db.Exec("UPDATE webdav_sync_state SET base_hash=?,conflict=0,local_hash='',remote_hash='',action=? WHERE id=1", hash, action)
	return err
}
func (s *Store) SetSyncConflict(local, remote string) error {
	_, err := s.db.Exec("UPDATE webdav_sync_state SET conflict=1,local_hash=?,remote_hash=?,action='conflict' WHERE id=1", local, remote)
	return err
}