♻️ Refactor synchronization service
This commit is contained in:
248
internal/syncer/snapshot/snapshot.go
Normal file
248
internal/syncer/snapshot/snapshot.go
Normal file
@@ -0,0 +1,248 @@
|
||||
package snapshot
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
// CurrentVersion 是当前快照格式版本。
|
||||
CurrentVersion = 1
|
||||
)
|
||||
|
||||
// Snapshot 描述一次完整的数据快照。
|
||||
type Snapshot struct {
|
||||
Version int
|
||||
CreatedAt time.Time
|
||||
Resources map[string][]Record
|
||||
}
|
||||
|
||||
// Record 描述单条资源记录。
|
||||
type Record struct {
|
||||
Kind string
|
||||
ID string
|
||||
UpdatedAt time.Time
|
||||
DeletedAt *time.Time
|
||||
Values map[string]interface{}
|
||||
Blobs map[string][]byte
|
||||
}
|
||||
|
||||
// Snapshotter 描述快照导出与应用接口。
|
||||
type Snapshotter interface {
|
||||
Export(ctx context.Context) (*Snapshot, error)
|
||||
Apply(ctx context.Context, snap *Snapshot) error
|
||||
}
|
||||
|
||||
// New 创建新的空快照。
|
||||
func New() *Snapshot {
|
||||
return &Snapshot{
|
||||
Version: CurrentVersion,
|
||||
CreatedAt: time.Now(),
|
||||
Resources: make(map[string][]Record),
|
||||
}
|
||||
}
|
||||
|
||||
// NewRecord 根据业务字段构造规范化记录。
|
||||
func NewRecord(kind string, id string, values map[string]interface{}, blobs map[string][]byte) (Record, error) {
|
||||
if strings.TrimSpace(kind) == "" {
|
||||
return Record{}, errors.New("record kind is required")
|
||||
}
|
||||
|
||||
normalizedValues := cloneValues(values)
|
||||
if id == "" {
|
||||
uuid, _ := normalizedValues["uuid"].(string)
|
||||
id = uuid
|
||||
}
|
||||
if id == "" {
|
||||
return Record{}, errors.New("record id is required")
|
||||
}
|
||||
normalizedValues["uuid"] = id
|
||||
|
||||
updatedAt, err := parseRequiredTime(normalizedValues["updated_at"])
|
||||
if err != nil {
|
||||
return Record{}, fmt.Errorf("record %s updated_at: %w", id, err)
|
||||
}
|
||||
|
||||
deletedAt, err := parseOptionalTime(normalizedValues["deleted_at"])
|
||||
if err != nil {
|
||||
return Record{}, fmt.Errorf("record %s deleted_at: %w", id, err)
|
||||
}
|
||||
|
||||
return Record{
|
||||
Kind: kind,
|
||||
ID: id,
|
||||
UpdatedAt: updatedAt,
|
||||
DeletedAt: deletedAt,
|
||||
Values: normalizedValues,
|
||||
Blobs: cloneBlobs(blobs),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Clone 返回快照的深拷贝。
|
||||
func Clone(snap *Snapshot) *Snapshot {
|
||||
if snap == nil {
|
||||
return New()
|
||||
}
|
||||
|
||||
cloned := &Snapshot{
|
||||
Version: snap.Version,
|
||||
CreatedAt: snap.CreatedAt,
|
||||
Resources: make(map[string][]Record, len(snap.Resources)),
|
||||
}
|
||||
for kind, records := range snap.Resources {
|
||||
copied := make([]Record, 0, len(records))
|
||||
for _, record := range records {
|
||||
copied = append(copied, CloneRecord(record))
|
||||
}
|
||||
cloned.Resources[kind] = copied
|
||||
}
|
||||
return cloned
|
||||
}
|
||||
|
||||
// CloneRecord 返回记录的深拷贝。
|
||||
func CloneRecord(record Record) Record {
|
||||
return Record{
|
||||
Kind: record.Kind,
|
||||
ID: record.ID,
|
||||
UpdatedAt: record.UpdatedAt,
|
||||
DeletedAt: cloneTime(record.DeletedAt),
|
||||
Values: cloneValues(record.Values),
|
||||
Blobs: cloneBlobs(record.Blobs),
|
||||
}
|
||||
}
|
||||
|
||||
// Digest 计算快照的稳定摘要。
|
||||
func Digest(snap *Snapshot) (string, error) {
|
||||
normalized := Clone(snap)
|
||||
|
||||
type digestRecord struct {
|
||||
ID string `json:"id"`
|
||||
UpdatedAt string `json:"updated_at"`
|
||||
DeletedAt *string `json:"deleted_at,omitempty"`
|
||||
Values map[string]interface{} `json:"values"`
|
||||
Blobs map[string][]byte `json:"blobs,omitempty"`
|
||||
}
|
||||
|
||||
payload := struct {
|
||||
Version int `json:"version"`
|
||||
Resources map[string][]digestRecord `json:"resources"`
|
||||
}{
|
||||
Version: normalized.Version,
|
||||
Resources: make(map[string][]digestRecord, len(normalized.Resources)),
|
||||
}
|
||||
|
||||
for _, kind := range sortedKinds(normalized.Resources) {
|
||||
records := normalized.Resources[kind]
|
||||
sort.Slice(records, func(i int, j int) bool {
|
||||
return records[i].ID < records[j].ID
|
||||
})
|
||||
|
||||
items := make([]digestRecord, 0, len(records))
|
||||
for _, record := range records {
|
||||
var deletedAt *string
|
||||
if record.DeletedAt != nil {
|
||||
value := record.DeletedAt.Format(time.RFC3339)
|
||||
deletedAt = &value
|
||||
}
|
||||
items = append(items, digestRecord{
|
||||
ID: record.ID,
|
||||
UpdatedAt: record.UpdatedAt.Format(time.RFC3339),
|
||||
DeletedAt: deletedAt,
|
||||
Values: record.Values,
|
||||
Blobs: record.Blobs,
|
||||
})
|
||||
}
|
||||
payload.Resources[kind] = items
|
||||
}
|
||||
|
||||
data, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
sum := sha256.Sum256(data)
|
||||
return hex.EncodeToString(sum[:]), nil
|
||||
}
|
||||
|
||||
// RecordDigest 计算单条记录的稳定摘要。
|
||||
func RecordDigest(record Record) string {
|
||||
sum, err := Digest(&Snapshot{
|
||||
Version: CurrentVersion,
|
||||
Resources: map[string][]Record{
|
||||
record.Kind: {CloneRecord(record)},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return sum
|
||||
}
|
||||
|
||||
// cloneValues 复制字段 map。
|
||||
func cloneValues(values map[string]interface{}) map[string]interface{} {
|
||||
if values == nil {
|
||||
return map[string]interface{}{}
|
||||
}
|
||||
return maps.Clone(values)
|
||||
}
|
||||
|
||||
// cloneBlobs 复制二进制 blob 集合。
|
||||
func cloneBlobs(blobs map[string][]byte) map[string][]byte {
|
||||
if len(blobs) == 0 {
|
||||
return nil
|
||||
}
|
||||
copied := make(map[string][]byte, len(blobs))
|
||||
for name, blob := range blobs {
|
||||
copied[name] = append([]byte(nil), blob...)
|
||||
}
|
||||
return copied
|
||||
}
|
||||
|
||||
// cloneTime 复制时间指针。
|
||||
func cloneTime(value *time.Time) *time.Time {
|
||||
if value == nil {
|
||||
return nil
|
||||
}
|
||||
cloned := *value
|
||||
return &cloned
|
||||
}
|
||||
|
||||
// parseRequiredTime 解析必填时间字段。
|
||||
func parseRequiredTime(value interface{}) (time.Time, error) {
|
||||
text, _ := value.(string)
|
||||
if text == "" {
|
||||
return time.Time{}, errors.New("time value is required")
|
||||
}
|
||||
return time.Parse(time.RFC3339, text)
|
||||
}
|
||||
|
||||
// parseOptionalTime 解析可选时间字段。
|
||||
func parseOptionalTime(value interface{}) (*time.Time, error) {
|
||||
text, _ := value.(string)
|
||||
if text == "" {
|
||||
return nil, nil
|
||||
}
|
||||
parsed, err := time.Parse(time.RFC3339, text)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &parsed, nil
|
||||
}
|
||||
|
||||
// sortedKinds 返回稳定排序后的资源类型列表。
|
||||
func sortedKinds(resources map[string][]Record) []string {
|
||||
kinds := make([]string, 0, len(resources))
|
||||
for kind := range resources {
|
||||
kinds = append(kinds, kind)
|
||||
}
|
||||
sort.Strings(kinds)
|
||||
return kinds
|
||||
}
|
||||
266
internal/syncer/snapshot/store.go
Normal file
266
internal/syncer/snapshot/store.go
Normal file
@@ -0,0 +1,266 @@
|
||||
package snapshot
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const manifestFileName = "manifest.json"
|
||||
|
||||
// Store 描述快照落盘与读取能力。
|
||||
type Store interface {
|
||||
Read(ctx context.Context, root string) (*Snapshot, error)
|
||||
Write(ctx context.Context, root string, snap *Snapshot) error
|
||||
}
|
||||
|
||||
// FileStore 提供基于目录树的快照读写实现。
|
||||
type FileStore struct{}
|
||||
|
||||
type manifest struct {
|
||||
Version int `json:"version"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
}
|
||||
|
||||
// NewFileStore 创建新的文件快照存储。
|
||||
func NewFileStore() *FileStore {
|
||||
return &FileStore{}
|
||||
}
|
||||
|
||||
// Read 从目录树读取快照。
|
||||
func (s *FileStore) Read(ctx context.Context, root string) (*Snapshot, error) {
|
||||
_ = ctx
|
||||
|
||||
info, err := os.Stat(root)
|
||||
if os.IsNotExist(err) {
|
||||
return New(), nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !info.IsDir() {
|
||||
return New(), nil
|
||||
}
|
||||
|
||||
snap := New()
|
||||
if err := s.readManifest(root, snap); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
entries, err := os.ReadDir(root)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, entry := range entries {
|
||||
if !entry.IsDir() {
|
||||
continue
|
||||
}
|
||||
|
||||
kind := entry.Name()
|
||||
records, err := s.readKind(filepath.Join(root, kind), kind)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(records) == 0 {
|
||||
continue
|
||||
}
|
||||
sort.Slice(records, func(i int, j int) bool {
|
||||
return records[i].ID < records[j].ID
|
||||
})
|
||||
snap.Resources[kind] = records
|
||||
}
|
||||
|
||||
return snap, nil
|
||||
}
|
||||
|
||||
// Write 将快照写入目录树。
|
||||
func (s *FileStore) Write(ctx context.Context, root string, snap *Snapshot) error {
|
||||
_ = ctx
|
||||
|
||||
if err := os.RemoveAll(root); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.MkdirAll(root, 0755); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := s.writeManifest(root, snap); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, kind := range sortedKinds(snap.Resources) {
|
||||
kindDir := filepath.Join(root, kind)
|
||||
if err := os.MkdirAll(kindDir, 0755); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
records := append([]Record(nil), snap.Resources[kind]...)
|
||||
sort.Slice(records, func(i int, j int) bool {
|
||||
return records[i].ID < records[j].ID
|
||||
})
|
||||
|
||||
for _, record := range records {
|
||||
if len(record.Blobs) == 0 {
|
||||
if err := writeJSON(filepath.Join(kindDir, record.ID+".json"), record.Values); err != nil {
|
||||
return err
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
recordDir := filepath.Join(kindDir, record.ID)
|
||||
if err := os.MkdirAll(recordDir, 0755); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := writeJSON(filepath.Join(recordDir, "record.json"), record.Values); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
blobNames := make([]string, 0, len(record.Blobs))
|
||||
for name := range record.Blobs {
|
||||
blobNames = append(blobNames, name)
|
||||
}
|
||||
sort.Strings(blobNames)
|
||||
|
||||
for _, blobName := range blobNames {
|
||||
if err := os.WriteFile(filepath.Join(recordDir, blobName), record.Blobs[blobName], 0644); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// readManifest 读取快照 manifest。
|
||||
func (s *FileStore) readManifest(root string, snap *Snapshot) error {
|
||||
data, err := os.ReadFile(filepath.Join(root, manifestFileName))
|
||||
if os.IsNotExist(err) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var current manifest
|
||||
if err := json.Unmarshal(data, ¤t); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
snap.Version = current.Version
|
||||
if current.CreatedAt != "" {
|
||||
createdAt, err := time.Parse(time.RFC3339, current.CreatedAt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
snap.CreatedAt = createdAt
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// writeManifest 写入快照 manifest。
|
||||
func (s *FileStore) writeManifest(root string, snap *Snapshot) error {
|
||||
payload := manifest{
|
||||
Version: snap.Version,
|
||||
CreatedAt: snap.CreatedAt.Format(time.RFC3339),
|
||||
}
|
||||
return writeJSON(filepath.Join(root, manifestFileName), payload)
|
||||
}
|
||||
|
||||
// readKind 读取单类资源目录。
|
||||
func (s *FileStore) readKind(root string, kind string) ([]Record, error) {
|
||||
entries, err := os.ReadDir(root)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
records := make([]Record, 0, len(entries))
|
||||
for _, entry := range entries {
|
||||
switch {
|
||||
case entry.IsDir():
|
||||
record, err := s.readBlobRecord(filepath.Join(root, entry.Name()), kind)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
records = append(records, record)
|
||||
case strings.HasSuffix(entry.Name(), ".json"):
|
||||
record, err := s.readFlatRecord(filepath.Join(root, entry.Name()), kind)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
records = append(records, record)
|
||||
}
|
||||
}
|
||||
return records, nil
|
||||
}
|
||||
|
||||
// readFlatRecord 读取单文件记录。
|
||||
func (s *FileStore) readFlatRecord(path string, kind string) (Record, error) {
|
||||
values, err := readValues(path)
|
||||
if err != nil {
|
||||
return Record{}, err
|
||||
}
|
||||
|
||||
id := strings.TrimSuffix(filepath.Base(path), filepath.Ext(path))
|
||||
return NewRecord(kind, id, values, nil)
|
||||
}
|
||||
|
||||
// readBlobRecord 读取目录型记录。
|
||||
func (s *FileStore) readBlobRecord(root string, kind string) (Record, error) {
|
||||
values, err := readValues(filepath.Join(root, "record.json"))
|
||||
if err != nil {
|
||||
return Record{}, err
|
||||
}
|
||||
|
||||
entries, err := os.ReadDir(root)
|
||||
if err != nil {
|
||||
return Record{}, err
|
||||
}
|
||||
|
||||
blobs := make(map[string][]byte)
|
||||
for _, entry := range entries {
|
||||
if entry.IsDir() || entry.Name() == "record.json" {
|
||||
continue
|
||||
}
|
||||
content, err := os.ReadFile(filepath.Join(root, entry.Name()))
|
||||
if err != nil {
|
||||
return Record{}, err
|
||||
}
|
||||
blobs[entry.Name()] = content
|
||||
}
|
||||
|
||||
return NewRecord(kind, filepath.Base(root), values, blobs)
|
||||
}
|
||||
|
||||
// readValues 读取 JSON 字段集合。
|
||||
func readValues(path string) (map[string]interface{}, error) {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
values := make(map[string]interface{})
|
||||
if err := json.Unmarshal(data, &values); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return values, nil
|
||||
}
|
||||
|
||||
// writeJSON 将结构体格式化写入 JSON 文件。
|
||||
func writeJSON(path string, value interface{}) error {
|
||||
data, err := json.MarshalIndent(value, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
data = append(data, '\n')
|
||||
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, data, 0644)
|
||||
}
|
||||
59
internal/syncer/snapshot/store_test.go
Normal file
59
internal/syncer/snapshot/store_test.go
Normal file
@@ -0,0 +1,59 @@
|
||||
package snapshot
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestFileStoreReadWrite 验证目录树快照可以稳定往返。
|
||||
func TestFileStoreReadWrite(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
|
||||
documentRecord, err := NewRecord("documents", "doc-1", map[string]interface{}{
|
||||
"uuid": "doc-1",
|
||||
"updated_at": time.Date(2026, 3, 29, 10, 0, 0, 0, time.UTC).Format(time.RFC3339),
|
||||
"title": "hello",
|
||||
}, map[string][]byte{
|
||||
"content.md": []byte("world"),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("build document record: %v", err)
|
||||
}
|
||||
|
||||
themeRecord, err := NewRecord("themes", "theme-1", map[string]interface{}{
|
||||
"uuid": "theme-1",
|
||||
"updated_at": time.Date(2026, 3, 29, 10, 1, 0, 0, time.UTC).Format(time.RFC3339),
|
||||
"name": "dark",
|
||||
}, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("build theme record: %v", err)
|
||||
}
|
||||
|
||||
snap := New()
|
||||
snap.Resources["documents"] = []Record{documentRecord}
|
||||
snap.Resources["themes"] = []Record{themeRecord}
|
||||
|
||||
store := NewFileStore()
|
||||
if err := store.Write(context.Background(), root, snap); err != nil {
|
||||
t.Fatalf("write snapshot: %v", err)
|
||||
}
|
||||
|
||||
loaded, err := store.Read(context.Background(), root)
|
||||
if err != nil {
|
||||
t.Fatalf("read snapshot: %v", err)
|
||||
}
|
||||
|
||||
originalDigest, err := Digest(snap)
|
||||
if err != nil {
|
||||
t.Fatalf("digest original snapshot: %v", err)
|
||||
}
|
||||
loadedDigest, err := Digest(loaded)
|
||||
if err != nil {
|
||||
t.Fatalf("digest loaded snapshot: %v", err)
|
||||
}
|
||||
|
||||
if originalDigest != loadedDigest {
|
||||
t.Fatalf("expected digests to match, got %s != %s", originalDigest, loadedDigest)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user