JCS-pub/client/internal/spacesyncer/execute_diff.go

package spacesyncer

import (
	"context"
	"io"
	"time"

	"gitlink.org.cn/cloudream/common/pkgs/logger"
	"gitlink.org.cn/cloudream/common/pkgs/trie"
	"gitlink.org.cn/cloudream/common/utils/math2"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch/exec"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2/parser"
	stgtypes "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/types"
	jcstypes "gitlink.org.cn/cloudream/jcs-pub/common/types"
)

func executeDiff(syncer *SpaceSyncer, task *task, mode *jcstypes.SpaceSyncModeDiff) {
	log := logger.WithField("Mod", logMod).WithField("TaskID", task.Task.TaskID)

	startTime := time.Now()
	log.Infof("begin full sync task")
	defer func() {
		log.Infof("full sync task finished, time: %v", time.Since(startTime))
	}()

	srcSpace := syncer.spaceMeta.Get(task.Task.SrcUserSpaceID)
	if srcSpace == nil {
		log.Warnf("src space %v not found", task.Task.SrcUserSpaceID)
		return
	}

	if len(task.Task.Dests) > 1 {
		log.Warnf("diff mode only support one dest now")
	}

	dstSpace := syncer.spaceMeta.Get(task.Task.Dests[0].DestUserSpaceID)
	if dstSpace == nil {
		log.Warnf("dest space %v not found", task.Task.Dests[0].DestUserSpaceID)
		return
	}

	srcBase, err := syncer.stgPool.GetBaseStore(srcSpace)
	if err != nil {
		log.Warnf("get src base store error: %v", err)
		return
	}

	dstBase, err := syncer.stgPool.GetBaseStore(dstSpace)
	if err != nil {
		log.Warnf("get dst base store error: %v", err)
		return
	}

	filter := buildFilter(task)

	srcReader := srcBase.ReadDir(task.Task.SrcPath)
	dstReader := dstBase.ReadDir(task.Task.Dests[0].DestPath)

	dirTree := trie.NewTrie[srcDstDirEntry]()

	for {
		e, err := srcReader.Next()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Warnf("read src dir: %v", err)
			return
		}

		if !filter(e) {
			continue
		}

		rela := e.Path.Clone()
		rela.PopFrontN(task.Task.SrcPath.Len())

		ne := e
		ne.Path = rela.Clone()
		if !filter(ne) {
			continue
		}

		diffCreateSrcNode(dirTree, rela, &e)
	}

	for {
		e, err := dstReader.Next()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Warnf("read dst dir: %v", err)
			return
		}

		if !filter(e) {
			continue
		}

		rela := e.Path.Clone()
		rela.PopFrontN(task.Task.Dests[0].DestPath.Len())

		ne := e
		ne.Path = rela.Clone()
		if !filter(ne) {
			continue
		}

		diffCreateDstNode(dirTree, rela, &e)
	}

	var willSync []stgtypes.DirEntry
	var willMkdirs []jcstypes.JPath

	dirTree.Iterate(func(path []string, node *trie.Node[srcDstDirEntry], isWordNode bool) trie.VisitCtrl {
		if node.Value.src == nil {
			// 目前不支持删除多余文件
			return trie.VisitContinue
		}

		if node.Value.src.IsDir {
			if node.Value.dst == nil {
				if node.IsEmpty() {
					willMkdirs = append(willMkdirs, jcstypes.PathFromComps(path...))
				}
			}
		} else {
			if node.Value.dst == nil {
				// 目标路径不存在（不是文件也不是目录），需要同步
				if node.IsEmpty() {
					willSync = append(willSync, *node.Value.src)
				}
			} else if !node.Value.dst.IsDir {
				// 目标路径是个文件，但文件指纹不同，需要同步
				if !cmpFile(mode, node.Value.src, node.Value.dst) {
					willSync = append(willSync, *node.Value.src)
				}
			}

			// 目标路径是个目录，则不进行同步
		}

		return trie.VisitContinue
	})

	willSyncCnt := len(willSync)
	for len(willSync) > 0 {
		syncs := willSync[:math2.Min(len(willSync), 50)]
		willSync = willSync[len(syncs):]

		ft := ioswitch2.NewFromTo()
		for _, s := range syncs {
			ft.AddFrom(ioswitch2.NewFromBaseStore(*srcSpace, s.Path))
			rela := s.Path.Clone()
			rela.PopFrontN(task.Task.SrcPath.Len())
			dstPath := task.Task.Dests[0].DestPath.ConcatNew(rela)
			to := ioswitch2.NewToBaseStore(*dstSpace, dstPath)
			to.Option.ModTime = s.ModTime
			ft.AddTo(to)
		}

		planBld := exec.NewPlanBuilder()
		err := parser.Parse(ft, planBld)
		if err != nil {
			log.Warnf("parse fromto: %v", err)
			return
		}

		execCtx := exec.NewWithContext(task.Context)
		exec.SetValueByType(execCtx, syncer.stgPool)
		_, err = planBld.Execute(execCtx).Wait(context.Background())
		if err != nil {
			log.Warnf("execute plan: %v", err)
			return
		}
	}

	log.Infof("%v files synced", willSyncCnt)

	if !task.Task.Options.NoEmptyDirectories && len(willMkdirs) > 0 {
		for _, p := range willMkdirs {
			rela := p.Clone()
			rela.PopFrontN(task.Task.SrcPath.Len())
			dstPath := task.Task.Dests[0].DestPath.ConcatNew(rela)
			err := dstBase.Mkdir(dstPath)
			if err != nil {
				log.Warnf("mkdir: %v", err)
				continue
			}
		}
	}
}

func diffCreateSrcNode(tree *trie.Trie[srcDstDirEntry], path jcstypes.JPath, e *stgtypes.DirEntry) {
	var ptr = &tree.Root
	for _, c := range path.Comps() {
		if ptr.Value.src != nil && ptr.Value.src.IsDir {
			ptr.Value.src = nil
		}
		ptr = ptr.Create(c)
	}

	ptr.Value.src = e
}

func diffCreateDstNode(tree *trie.Trie[srcDstDirEntry], path jcstypes.JPath, e *stgtypes.DirEntry) {
	var ptr = &tree.Root
	for _, c := range path.Comps() {
		if ptr.Value.src != nil && ptr.Value.src.IsDir {
			ptr.Value.src = nil
		}

		if ptr.Value.dst != nil && ptr.Value.dst.IsDir {
			ptr.Value.dst = nil
		}

		ptr = ptr.Create(c)
	}

	ptr.Value.dst = e
}

type srcDstDirEntry struct {
	src *stgtypes.DirEntry
	dst *stgtypes.DirEntry
}

func cmpFile(diff *jcstypes.SpaceSyncModeDiff, src, dst *stgtypes.DirEntry) bool {
	if diff.IncludeSize && src.Size != dst.Size {
		return false
	}

	if diff.IncludeModTime && src.ModTime != dst.ModTime {
		return false
	}

	return true
}