// Package xfer implements haul's SFTP file-transfer engine: recursive push/pull // with aggregate progress, per-file overwrite policy, and same-session resume. // // Push (local→remote) and pull (remote→local) share one recursive copy routine // by abstracting each side behind vfs.FS, so the tree-walking, resume, and // conflict logic is written once and exercised in both directions. package xfer import ( "errors" "fmt" "io" "os" "sync/atomic" "github.com/pkg/sftp" "gitea.apointless.space/bsncubed/haul/internal/vfs" ) // ErrCanceled is returned by Push/Pull when Cancel was called mid-transfer. // Files already copied are left in place; the partial file is left on disk so // that re-running the same transfer on this Engine resumes it. var ErrCanceled = errors.New("transfer canceled") // Conflict is the decision for a destination file that already existed before // this transfer began. type Conflict int const ( Overwrite Conflict = iota // replace the existing file Skip // leave the existing file, skip the source Rename // write the source under a non-colliding name ) // Progress is a snapshot of an in-flight transfer, reported to OnProgress. type Progress struct { CurrentName string // file currently copying FilesDone int FilesTotal int BytesDone int64 // cumulative across all files BytesTotal int64 } // ConflictFunc decides what to do about a pre-existing destination file. name // is the destination path. Returning the same decision for every call is the // "apply to all" behaviour. type ConflictFunc func(name string) Conflict // ProgressFunc receives periodic progress updates. It must not block. type ProgressFunc func(Progress) // Engine performs transfers over one SFTP connection. A single Engine instance // remembers which destination files it has created this session, which is what // makes same-session resume work: re-running an interrupted transfer resumes // partial files it started rather than treating them as pre-existing conflicts. type Engine struct { remote vfs.FS OnProgress ProgressFunc OnConflict ConflictFunc created map[string]bool // destination paths this session started/finished canceled atomic.Bool // set by Cancel, cleared at the start of each run } // New returns an Engine bound to an SFTP client. func New(c *sftp.Client) *Engine { return &Engine{remote: vfs.Remote{C: c}, created: map[string]bool{}} } // Cancel asks an in-flight transfer to stop. It is safe to call from another // goroutine (the GUI's stop button) and takes effect at the next file or write // boundary; Push/Pull then return ErrCanceled. func (e *Engine) Cancel() { e.canceled.Store(true) } // Push copies local sources into the remote directory (recursively for dirs). func (e *Engine) Push(localSources []string, remoteDir string) error { return e.run(vfs.Local{}, localSources, e.remote, remoteDir) } // Pull copies remote sources into the local directory (recursively for dirs). func (e *Engine) Pull(remoteSources []string, localDir string) error { return e.run(e.remote, remoteSources, vfs.Local{}, localDir) } func (e *Engine) run(src vfs.FS, sources []string, dst vfs.FS, dstDir string) error { e.canceled.Store(false) prog := Progress{} // First pass: count files and total bytes for aggregate progress. for _, s := range sources { n, bytes, err := countTree(src, s) if err != nil { return err } prog.FilesTotal += n prog.BytesTotal += bytes } if err := dst.MkdirAll(dstDir); err != nil { return err } for _, s := range sources { target := dst.Join(dstDir, src.Base(s)) if err := e.copyTree(src, s, dst, target, &prog); err != nil { return err } } return nil } // countTree returns the file count and total byte size under p. func countTree(src vfs.FS, p string) (int, int64, error) { fi, err := src.Stat(p) if err != nil { return 0, 0, err } if !fi.IsDir() { return 1, fi.Size(), nil } entries, err := src.ReadDir(p) if err != nil { return 0, 0, err } var files int var bytes int64 for _, ent := range entries { n, b, err := countTree(src, src.Join(p, ent.Name())) if err != nil { return 0, 0, err } files += n bytes += b } return files, bytes, nil } func (e *Engine) copyTree(src vfs.FS, srcPath string, dst vfs.FS, dstPath string, prog *Progress) error { if e.canceled.Load() { return ErrCanceled } fi, err := src.Stat(srcPath) if err != nil { return err } if fi.IsDir() { if err := dst.MkdirAll(dstPath); err != nil { return err } entries, err := src.ReadDir(srcPath) if err != nil { return err } for _, ent := range entries { child := dst.Join(dstPath, ent.Name()) if err := e.copyTree(src, src.Join(srcPath, ent.Name()), dst, child, prog); err != nil { return err } } return nil } return e.copyFile(src, srcPath, fi.Size(), dst, dstPath, prog) } // copyFile copies a single regular file, honouring conflict policy for // pre-existing destinations and resuming partials this session started. func (e *Engine) copyFile(src vfs.FS, srcPath string, srcSize int64, dst vfs.FS, dstPath string, prog *Progress) error { var offset int64 if info, err := dst.Stat(dstPath); err == nil { if e.created[dstPath] { // Our own partial from an interrupted attempt: resume from its end. if info.Size() <= srcSize { offset = info.Size() } } else { // Pre-existing file: ask the caller what to do. switch e.conflict(dstPath) { case Skip: prog.FilesDone++ prog.BytesDone += srcSize e.report(prog, dst.Base(dstPath)) return nil case Rename: dstPath = uniqueName(dst, dstPath) case Overwrite: // fall through; Create truncates. } } } in, err := src.Open(srcPath) if err != nil { return err } defer in.Close() var out io.WriteCloser if offset > 0 { if _, err := in.Seek(offset, io.SeekStart); err != nil { return err } if out, err = dst.OpenAt(dstPath, offset); err != nil { return err } } else if out, err = dst.Create(dstPath); err != nil { return err } e.created[dstPath] = true prog.BytesDone += offset // resumed bytes already count as done pw := &progressWriter{w: out, prog: prog, name: dst.Base(dstPath), report: e.report, canceled: &e.canceled} if _, err := io.Copy(pw, in); err != nil { out.Close() return fmt.Errorf("copying %s: %w", srcPath, err) } if err := out.Close(); err != nil { return err } prog.FilesDone++ e.report(prog, dst.Base(dstPath)) return nil } func (e *Engine) conflict(name string) Conflict { if e.OnConflict == nil { return Overwrite } return e.OnConflict(name) } func (e *Engine) report(prog *Progress, name string) { if e.OnProgress == nil { return } p := *prog p.CurrentName = name e.OnProgress(p) } // uniqueName returns dstPath with a numeric suffix that doesn't yet exist. func uniqueName(dst vfs.FS, dstPath string) string { for i := 1; ; i++ { cand := fmt.Sprintf("%s.%d", dstPath, i) if _, err := dst.Stat(cand); errors.Is(err, os.ErrNotExist) || err != nil { return cand } } } // progressWriter tallies bytes as they're written and emits progress updates. // It is also where a cancel lands mid-file, so a single large file can be // interrupted rather than only being stoppable between files. type progressWriter struct { w io.Writer prog *Progress name string report func(*Progress, string) canceled *atomic.Bool } func (p *progressWriter) Write(b []byte) (int, error) { if p.canceled != nil && p.canceled.Load() { return 0, ErrCanceled } n, err := p.w.Write(b) p.prog.BytesDone += int64(n) p.report(p.prog, p.name) return n, err }