package syncengine import ( "context" "fmt" "io" "os/exec" "strings" ) type RsyncResult struct { ExitCode int Stdout string Stderr string Stats *RsyncStats } type RsyncRunner struct { sshDir string privKey string } type SyncPairConfig struct { ID int64 Name string SourceMachineID *int64 SourcePath string DestMachineID *int64 DestPath string Direction string RsyncFlags string ExcludePatterns []string Source string Dest string } func NewRsyncRunner(sshDir, privKey string) *RsyncRunner { return &RsyncRunner{sshDir: sshDir, privKey: privKey} } func (r *RsyncRunner) Run(ctx context.Context, pair *SyncPairConfig, onLine func(stream string, line string)) (*RsyncResult, error) { args := r.buildArgs(pair) cmd := exec.CommandContext(ctx, "rsync", args...) if r.privKey != "" { sshCmd := fmt.Sprintf("ssh -i %s -o StrictHostKeyChecking=accept-new -o UserKnownHostsFile=%s", r.privKey, strings.TrimRight(r.sshDir, "/")+"/known_hosts") cmd.Args = append([]string{"rsync", "-e", sshCmd}, args[1:]...) } stdout, err := cmd.StdoutPipe() if err != nil { return nil, fmt.Errorf("stdout pipe: %w", err) } stderr, err := cmd.StderrPipe() if err != nil { return nil, fmt.Errorf("stderr pipe: %w", err) } if err := cmd.Start(); err != nil { return nil, fmt.Errorf("starting rsync: %w", err) } var outLines, errLines []string done := make(chan struct{}) go func() { br := io.Reader(stdout) buf := make([]byte, 4096) for { n, err := br.Read(buf) if n > 0 { line := strings.TrimRight(string(buf[:n]), "\r\n") if line != "" { outLines = append(outLines, line) if onLine != nil { onLine("stdout", line) } } } if err != nil { break } } select { case <-done: default: close(done) } }() go func() { br := io.Reader(stderr) buf := make([]byte, 4096) for { n, err := br.Read(buf) if n > 0 { line := strings.TrimRight(string(buf[:n]), "\r\n") if line != "" { errLines = append(errLines, line) if onLine != nil { onLine("stderr", line) } } } if err != nil { break } } select { case <-done: default: close(done) } }() err = cmd.Wait() <-done result := &RsyncResult{ ExitCode: 0, Stdout: strings.Join(outLines, "\n"), Stderr: strings.Join(errLines, "\n"), Stats: ParseFinalStats(strings.Join(outLines, "\n")), } if err != nil { if exitErr, ok := err.(*exec.ExitError); ok { result.ExitCode = exitErr.ExitCode() } else { result.ExitCode = -1 } } return result, nil } func (r *RsyncRunner) buildArgs(pair *SyncPairConfig) []string { var args []string flags := strings.Fields(pair.RsyncFlags) args = append(args, flags...) for _, pattern := range pair.ExcludePatterns { args = append(args, "--exclude="+pattern) } if pair.Direction == "mirror" { args = append(args, "--delete") } if pair.Direction == "pull" { args = append(args, pair.Dest, pair.Source) } else { args = append(args, pair.Source, pair.Dest) } return args }