haul/internal/xfer/xfer.go

// 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
}