goget/internal/protocol/sftp/sftp.go

//go:build linux || freebsd
// +build linux freebsd

// Package sftp implements the SFTP (SSH File Transfer Protocol) version 3.
package sftp

import (
	"bytes"
	"context"
	"encoding/binary"
	"fmt"
	"io"
	"net"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"codeberg.org/petrbalvin/goget/internal/core"
	"codeberg.org/petrbalvin/goget/internal/output"
	"codeberg.org/petrbalvin/goget/internal/protocol"
	"golang.org/x/crypto/ssh"
)

// sftp packet types
const (
	sshFxpInit     = 1
	sshFxpVersion  = 2
	sshFxpOpen     = 3
	sshFxpClose    = 4
	sshFxpRead     = 5
	sshFxpWrite    = 6
	sshFxpLstat    = 7
	sshFxpFstat    = 8
	sshFxpSetstat  = 9
	sshFxpFsetstat = 10
	sshFxpOpendir  = 11
	sshFxpReaddir  = 12
	sshFxpRemove   = 13
	sshFxpMkdir    = 14
	sshFxpRmdir    = 15
	sshFxpRealpath = 16
	sshFxpStat     = 17
	sshFxpRename   = 18
	sshFxpReadlink = 19
	sshFxpSymlink  = 20

	sshFxpStatus = 101
	sshFxpHandle = 102
	sshFxpData   = 103
	sshFxpName   = 104
	sshFxpAttrs  = 105

	sshFxOk               = 0
	sshFxEOF              = 1
	sshFxNoSuchFile       = 2
	sshFxPermissionDenied = 3
	sshFxFailure          = 4
	sshFxBadMessage       = 5
	sshFxNoConnection     = 6
	sshFxConnectionLost   = 7
	sshFxOpUnsupported    = 8

	sshFxfRead   = 0x00000001
	sshFxfWrite  = 0x00000002
	sshFxfAppend = 0x00000004
	sshFxfCreat  = 0x00000008
	sshFxfTrunc  = 0x00000010
	sshFxfExcl   = 0x00000020
)

// Protocol implements SFTP downloads via SSH.
// Supports single-file downloads, recursive directory downloads,
// and parallel recursive downloads via worker-pool + connection pooling.
type Protocol struct {
	protocol.BaseProtocol
}

// NewProtocol creates a new SFTP protocol handler.
func NewProtocol() *Protocol {
	return &Protocol{
		BaseProtocol: *protocol.NewBaseProtocol(protocol.ProtocolInfo{
			Name:        "sftp",
			Scheme:      "sftp",
			Version:     "3",
			Operations:  []string{"download", "upload"},
			Features:    []string{"resume", "recursive", "parallel"},
			DefaultPort: 22,
		}),
	}
}

func (p *Protocol) CanHandle(u *url.URL) bool {
	if u == nil {
		return false
	}
	return u.Scheme == "sftp"
}

func (p *Protocol) Download(ctx context.Context, req *core.DownloadRequest) (*core.DownloadResult, error) {
	return download(ctx, req)
}

// client wraps an SSH connection and SFTP session.
type client struct {
	sshConn   *ssh.Client
	sftpChan  ssh.Channel
	reqID     uint32
	reqIDMu   sync.Mutex
	maxPacket uint32
}

func dial(ctx context.Context, u *url.URL, knownHostsPath string, insecure bool) (*client, error) {
	host := u.Hostname()
	port := u.Port()
	if port == "" {
		port = "22"
	}

	user := u.User.Username()
	if user == "" {
		user = "root"
	}

	addr := net.JoinHostPort(host, port)

	hostKeyCB, err := hostKeyCallback(knownHostsPath, insecure)
	if err != nil {
		return nil, err
	}

	config := &ssh.ClientConfig{
		User:            user,
		HostKeyCallback: hostKeyCB,
		Timeout:         30 * time.Second,
	}

	if pass, ok := u.User.Password(); ok {
		config.Auth = []ssh.AuthMethod{
			ssh.Password(pass),
		}
	} else {
		// Try default SSH keys
		config.Auth = []ssh.AuthMethod{
			ssh.PublicKeysCallback(func() ([]ssh.Signer, error) {
				return defaultSigners()
			}),
		}
	}

	sshConn, err := ssh.Dial("tcp", addr, config)
	if err != nil {
		return nil, fmt.Errorf("ssh dial: %w", err)
	}

	session, err := sshConn.NewSession()
	if err != nil {
		sshConn.Close()
		return nil, fmt.Errorf("ssh session: %w", err)
	}

	stdin, err := session.StdinPipe()
	if err != nil {
		sshConn.Close()
		return nil, fmt.Errorf("stdin pipe: %w", err)
	}
	stdout, err := session.StdoutPipe()
	if err != nil {
		sshConn.Close()
		return nil, fmt.Errorf("stdout pipe: %w", err)
	}

	if err := session.RequestSubsystem("sftp"); err != nil {
		sshConn.Close()
		return nil, fmt.Errorf("sftp subsystem: %w", err)
	}

	c := &client{
		sshConn: sshConn,
		sftpChan: &sftpChannel{
			stdin:  stdin,
			stdout: stdout,
		},
		maxPacket: 32768,
	}

	if err := c.init(); err != nil {
		c.close()
		return nil, err
	}

	return c, nil
}

func (c *client) close() {
	if c.sftpChan != nil {
		c.sftpChan.Close()
	}
	if c.sshConn != nil {
		c.sshConn.Close()
	}
}

func (c *client) nextReqID() uint32 {
	c.reqIDMu.Lock()
	defer c.reqIDMu.Unlock()
	c.reqID++
	return c.reqID
}

// init performs the SFTP version handshake.
func (c *client) init() error {
	var buf bytes.Buffer
	buf.WriteByte(sshFxpInit)
	writeUint32(&buf, 3) // version 3

	if err := c.writePacket(buf.Bytes()); err != nil {
		return fmt.Errorf("sftp init write: %w", err)
	}

	// read response
	pkt, err := c.readPacket()
	if err != nil {
		return fmt.Errorf("sftp init read: %w", err)
	}
	if len(pkt) < 1 || pkt[0] != sshFxpVersion {
		return fmt.Errorf("unexpected init response: %d", pkt[0])
	}
	return nil
}

// readPacket reads a single SFTP packet.
func (c *client) readPacket() ([]byte, error) {
	var lenBuf [4]byte
	if _, err := io.ReadFull(c.sftpChan, lenBuf[:]); err != nil {
		return nil, err
	}
	length := binary.BigEndian.Uint32(lenBuf[:])
	if length > 256*1024 {
		return nil, fmt.Errorf("packet too large: %d", length)
	}
	pkt := make([]byte, length)
	if _, err := io.ReadFull(c.sftpChan, pkt); err != nil {
		return nil, err
	}
	return pkt, nil
}

// writePacket sends a single SFTP packet.
func (c *client) writePacket(pkt []byte) error {
	var lenBuf [4]byte
	binary.BigEndian.PutUint32(lenBuf[:], uint32(len(pkt)))
	if _, err := c.sftpChan.Write(lenBuf[:]); err != nil {
		return err
	}
	_, err := c.sftpChan.Write(pkt)
	return err
}

// sendRequest sends a request packet and returns the response.
func (c *client) sendRequest(pkt []byte) ([]byte, error) {
	if err := c.writePacket(pkt); err != nil {
		return nil, err
	}
	return c.readPacket()
}

// stat returns file attributes.
func (c *client) stat(path string) (*fileAttr, error) {
	reqID := c.nextReqID()
	var buf bytes.Buffer
	buf.WriteByte(sshFxpStat)
	writeUint32(&buf, reqID)
	writeString(&buf, path)

	resp, err := c.sendRequest(buf.Bytes())
	if err != nil {
		return nil, err
	}
	return parseStatusOrAttrs(resp)
}

// open opens a file for reading and returns a handle.
func (c *client) open(path string) (string, error) {
	reqID := c.nextReqID()
	var buf bytes.Buffer
	buf.WriteByte(sshFxpOpen)
	writeUint32(&buf, reqID)
	writeString(&buf, path)
	writeUint32(&buf, sshFxfRead) // pflags
	writeUint32(&buf, 0)          // attr flags (empty)

	resp, err := c.sendRequest(buf.Bytes())
	if err != nil {
		return "", err
	}
	return parseStatusOrHandle(resp)
}

// closeHandle closes a file/directory handle.
func (c *client) closeHandle(handle string) error {
	reqID := c.nextReqID()
	var buf bytes.Buffer
	buf.WriteByte(sshFxpClose)
	writeUint32(&buf, reqID)
	writeString(&buf, handle)

	resp, err := c.sendRequest(buf.Bytes())
	if err != nil {
		return err
	}
	return parseStatus(resp)
}

// read reads data from a file handle.
func (c *client) read(handle string, offset uint64, length uint32) ([]byte, error) {
	reqID := c.nextReqID()
	var buf bytes.Buffer
	buf.WriteByte(sshFxpRead)
	writeUint32(&buf, reqID)
	writeString(&buf, handle)
	writeUint64(&buf, offset)
	writeUint32(&buf, length)

	resp, err := c.sendRequest(buf.Bytes())
	if err != nil {
		return nil, err
	}
	return parseStatusOrData(resp)
}

// realpath resolves a path to an absolute path.
func (c *client) realpath(path string) (string, error) {
	reqID := c.nextReqID()
	var buf bytes.Buffer
	buf.WriteByte(sshFxpRealpath)
	writeUint32(&buf, reqID)
	writeString(&buf, path)

	resp, err := c.sendRequest(buf.Bytes())
	if err != nil {
		return "", err
	}
	return parseStatusOrName(resp)
}

// opendir opens a directory.
func (c *client) opendir(path string) (string, error) {
	reqID := c.nextReqID()
	var buf bytes.Buffer
	buf.WriteByte(sshFxpOpendir)
	writeUint32(&buf, reqID)
	writeString(&buf, path)

	resp, err := c.sendRequest(buf.Bytes())
	if err != nil {
		return "", err
	}
	return parseStatusOrHandle(resp)
}

// readdir reads directory entries.
func (c *client) readdir(handle string) ([]*dirEntry, error) {
	reqID := c.nextReqID()
	var buf bytes.Buffer
	buf.WriteByte(sshFxpReaddir)
	writeUint32(&buf, reqID)
	writeString(&buf, handle)

	resp, err := c.sendRequest(buf.Bytes())
	if err != nil {
		return nil, err
	}
	return parseStatusOrNames(resp)
}

// mkdir creates a directory.
func (c *client) mkdir(path string) error {
	reqID := c.nextReqID()
	var buf bytes.Buffer
	buf.WriteByte(sshFxpMkdir)
	writeUint32(&buf, reqID)
	writeString(&buf, path)
	writeUint32(&buf, 0) // attrs

	resp, err := c.sendRequest(buf.Bytes())
	if err != nil {
		return err
	}
	return parseStatus(resp)
}

// fileAttr holds file attributes.
type fileAttr struct {
	Size  uint64
	Mode  uint32
	UID   uint32
	GID   uint32
	Atime uint32
	Mtime uint32
}

type dirEntry struct {
	Name     string
	LongName string
	Attr     *fileAttr
}

// helpers
func writeUint32(w *bytes.Buffer, v uint32) {
	var b [4]byte
	binary.BigEndian.PutUint32(b[:], v)
	w.Write(b[:])
}

func writeUint64(w *bytes.Buffer, v uint64) {
	var b [8]byte
	binary.BigEndian.PutUint64(b[:], v)
	w.Write(b[:])
}

func writeString(w *bytes.Buffer, s string) {
	writeUint32(w, uint32(len(s)))
	w.WriteString(s)
}

func readUint32(b []byte, off int) (uint32, int) {
	return binary.BigEndian.Uint32(b[off:]), off + 4
}

func readUint64(b []byte, off int) (uint64, int) {
	return binary.BigEndian.Uint64(b[off:]), off + 8
}

func readString(b []byte, off int) (string, int) {
	l, off := readUint32(b, off)
	return string(b[off : off+int(l)]), off + int(l)
}

func parseStatusOrAttrs(pkt []byte) (*fileAttr, error) {
	if len(pkt) < 1 {
		return nil, fmt.Errorf("empty packet")
	}
	if pkt[0] == sshFxpStatus {
		return nil, parseStatus(pkt)
	}
	if pkt[0] != sshFxpAttrs {
		return nil, fmt.Errorf("unexpected packet type: %d", pkt[0])
	}
	_, off := readUint32(pkt, 1) // reqID
	attr, _ := readAttrs(pkt, off)
	return attr, nil
}

func parseStatusOrHandle(pkt []byte) (string, error) {
	if len(pkt) < 1 {
		return "", fmt.Errorf("empty packet")
	}
	if pkt[0] == sshFxpStatus {
		return "", parseStatus(pkt)
	}
	if pkt[0] != sshFxpHandle {
		return "", fmt.Errorf("unexpected packet type: %d", pkt[0])
	}
	_, off := readUint32(pkt, 1) // reqID
	handle, _ := readString(pkt, off)
	return handle, nil
}

func parseStatusOrData(pkt []byte) ([]byte, error) {
	if len(pkt) < 1 {
		return nil, fmt.Errorf("empty packet")
	}
	if pkt[0] == sshFxpStatus {
		if err := parseStatus(pkt); err != nil {
			if isEOF(err) {
				return nil, io.EOF
			}
			return nil, err
		}
		return nil, io.EOF
	}
	if pkt[0] != sshFxpData {
		return nil, fmt.Errorf("unexpected packet type: %d", pkt[0])
	}
	_, off := readUint32(pkt, 1) // reqID
	l, off := readUint32(pkt, off)
	return pkt[off : off+int(l)], nil
}

func parseStatusOrName(pkt []byte) (string, error) {
	if len(pkt) < 1 {
		return "", fmt.Errorf("empty packet")
	}
	if pkt[0] == sshFxpStatus {
		return "", parseStatus(pkt)
	}
	if pkt[0] != sshFxpName {
		return "", fmt.Errorf("unexpected packet type: %d", pkt[0])
	}
	_, off := readUint32(pkt, 1) // reqID
	count, off := readUint32(pkt, off)
	if count == 0 {
		return "", fmt.Errorf("no names in response")
	}
	name, _ := readString(pkt, off)
	return name, nil
}

func parseStatusOrNames(pkt []byte) ([]*dirEntry, error) {
	if len(pkt) < 1 {
		return nil, fmt.Errorf("empty packet")
	}
	if pkt[0] == sshFxpStatus {
		if err := parseStatus(pkt); err != nil {
			if isEOF(err) {
				return nil, io.EOF
			}
			return nil, err
		}
		return nil, io.EOF
	}
	if pkt[0] != sshFxpName {
		return nil, fmt.Errorf("unexpected packet type: %d", pkt[0])
	}
	_, off := readUint32(pkt, 1) // reqID
	count, off := readUint32(pkt, off)
	entries := make([]*dirEntry, 0, count)
	for i := uint32(0); i < count; i++ {
		var name, longName string
		name, off = readString(pkt, off)
		longName, off = readString(pkt, off)
		var attr *fileAttr
		attr, off = readAttrs(pkt, off)
		_ = longName
		entries = append(entries, &dirEntry{Name: name, LongName: name, Attr: attr})
	}
	return entries, nil
}

func parseStatus(pkt []byte) error {
	if len(pkt) < 5 {
		return fmt.Errorf("short status packet")
	}
	_, off := readUint32(pkt, 1) // reqID
	code, off := readUint32(pkt, off)
	msg, _ := readString(pkt, off)
	if code == sshFxOk {
		return nil
	}
	return fmt.Errorf("sftp status %d: %s", code, msg)
}

func isEOF(err error) bool {
	if err == nil {
		return false
	}
	return strings.Contains(err.Error(), fmt.Sprintf("sftp status %d", sshFxEOF))
}

func readAttrs(b []byte, off int) (*fileAttr, int) {
	flags, off := readUint32(b, off)
	attr := &fileAttr{}
	if flags&0x00000001 != 0 {
		attr.Size, off = readUint64(b, off)
	}
	if flags&0x00000002 != 0 {
		attr.UID, off = readUint32(b, off)
		attr.GID, off = readUint32(b, off)
	}
	if flags&0x00000004 != 0 {
		attr.Mode, off = readUint32(b, off)
	}
	if flags&0x00000008 != 0 {
		attr.Atime, off = readUint32(b, off)
		attr.Mtime, off = readUint32(b, off)
	}
	return attr, off
}

// defaultSigners loads default SSH private keys from ~/.ssh.
func defaultSigners() ([]ssh.Signer, error) {
	home, err := os.UserHomeDir()
	if err != nil {
		return nil, err
	}
	var signers []ssh.Signer
	for _, name := range []string{"id_ed25519", "id_rsa", "id_ecdsa"} {
		path := filepath.Join(home, ".ssh", name)
		data, err := os.ReadFile(path)
		if err != nil {
			continue
		}
		signer, err := ssh.ParsePrivateKey(data)
		if err != nil {
			continue
		}
		signers = append(signers, signer)
	}
	return signers, nil
}

// sftpChannel wraps stdin/stdout as an ssh.Channel.
type sftpChannel struct {
	stdin  io.WriteCloser
	stdout io.Reader
}

func (ch *sftpChannel) Read(p []byte) (int, error)  { return ch.stdout.Read(p) }
func (ch *sftpChannel) Write(p []byte) (int, error) { return ch.stdin.Write(p) }
func (ch *sftpChannel) Close() error {
	ch.stdin.Close()
	return nil
}
func (ch *sftpChannel) CloseWrite() error { return ch.stdin.Close() }
func (ch *sftpChannel) SendRequest(string, bool, []byte) (bool, error) {
	return false, nil
}
func (ch *sftpChannel) Stderr() io.ReadWriter { return &noopReadWriter{} }

type noopReadWriter struct{}

func (n *noopReadWriter) Read(p []byte) (int, error)  { return 0, io.EOF }
func (n *noopReadWriter) Write(p []byte) (int, error) { return len(p), nil }

// download performs the SFTP download. Dispatches to the single-file
// path or to the recursive path based on req.Recursive. The recursive
// path in turn dispatches to the sequential (single-connection) or
// parallel (worker-pool) implementation based on
// req.RecursiveParallel.
func download(ctx context.Context, req *core.DownloadRequest) (*core.DownloadResult, error) {
	if req.Recursive {
		if req.RecursiveParallel > 1 {
			return downloadRecursiveParallel(ctx, req)
		}
		return downloadRecursiveSequential(ctx, req)
	}

	c, err := dial(ctx, req.URL, req.SSHKnownHosts, req.SSHInsecure)
	if err != nil {
		return nil, err
	}
	defer c.close()

	return downloadSingleFile(ctx, c, req)
}

// downloadSingleFile downloads a single file using a pre-existing
// client. The caller is responsible for dialing and closing c.
func downloadSingleFile(ctx context.Context, c *client, req *core.DownloadRequest) (*core.DownloadResult, error) {
	remotePath := req.URL.Path
	if remotePath == "" {
		remotePath = "."
	}

	absPath, err := c.realpath(remotePath)
	if err != nil {
		absPath = remotePath
	}

	attr, err := c.stat(absPath)
	if err != nil {
		return nil, fmt.Errorf("sftp stat: %w", err)
	}

	outputPath := req.Output
	if outputPath == "" {
		outputPath = filepath.Base(absPath)
	}

	var startOffset int64
	resumed := false
	if req.Resume && outputPath != "" && outputPath != "-" {
		canResume, meta, err := output.CanResume(outputPath, req.URL.String())
		if err == nil && canResume && meta != nil {
			startOffset = meta.Downloaded
			resumed = true
		}
	}

	var writer io.Writer
	if outputPath == "-" {
		writer = os.Stdout
	} else {
		if err := os.MkdirAll(filepath.Dir(outputPath), 0755); err != nil {
			return nil, fmt.Errorf("mkdir output directory: %w", err)
		}
		f, err := os.OpenFile(outputPath, os.O_CREATE|os.O_WRONLY, 0644)
		if err != nil {
			return nil, fmt.Errorf("open output file: %w", err)
		}
		if startOffset > 0 {
			if _, err := f.Seek(startOffset, io.SeekStart); err != nil {
				f.Close()
				return nil, fmt.Errorf("seek output file: %w", err)
			}
		} else if err := f.Truncate(0); err != nil {
			f.Close()
			return nil, fmt.Errorf("truncate output file: %w", err)
		}
		defer f.Close()
		writer = f
	}

	startTime := time.Now()
	handle, err := c.open(absPath)
	if err != nil {
		return nil, fmt.Errorf("sftp open: %w", err)
	}
	defer c.closeHandle(handle)

	var total uint64
	offset := uint64(startOffset)
	if startOffset > 0 {
		total = uint64(startOffset)
	}
	bufSize := c.maxPacket
	if bufSize > 32768 {
		bufSize = 32768
	}
	totalSize := int64(attr.Size)

	for {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		default:
		}

		data, err := c.read(handle, offset, bufSize)
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("sftp read: %w", err)
		}

		n, werr := writer.Write(data)
		if werr != nil {
			return nil, fmt.Errorf("write output: %w", werr)
		}
		total += uint64(n)
		offset += uint64(n)

		if req.ProgressCallback != nil && totalSize > 0 {
			req.ProgressCallback(int64(total), totalSize, 0)
		}

		if outputPath != "" && outputPath != "-" && req.Resume {
			meta := output.NewResumeMetadata(req.URL.String(), "", "", int64(total), totalSize)
			_ = meta.Save(outputPath)
		}

		if uint64(n) < uint64(len(data)) {
			break
		}
	}

	if outputPath != "" && outputPath != "-" {
		_ = output.DeleteResumeMetadata(outputPath)
	}

	duration := time.Since(startTime)
	speed := float64(0)
	if duration.Seconds() > 0 {
		speed = float64(int64(total)-startOffset) / duration.Seconds()
	}
	result := &core.DownloadResult{
		BytesDownloaded: int64(total),
		Duration:        duration,
		Speed:           speed,
		OutputPath:      outputPath,
		Protocol:        "sftp",
		Resumed:         resumed,
	}
	// Note: do NOT overwrite BytesDownloaded with attr.Size. attr.Size is
	// the server-reported size, which may be stale, wrong (e.g. sparse
	// files), or larger than the bytes actually transferred on a
	// resumed/incomplete download. The locally-tracked `total` is the
	// accurate count of bytes the caller received.
	return result, nil
}

// sftpIsDir reports whether the SFTP mode indicates a directory.
// S_IFMT = 0xF000, S_IFDIR = 0x4000 (POSIX file mode bits).
func sftpIsDir(mode uint32) bool {
	return mode&0xF000 == 0x4000
}

// sftpClientPool holds N pre-dialed SFTP clients. A single shared pool
// is required for the parallel recursive implementation because each
// *client owns its own SSH+SFTP session and is not safe for
// concurrent use.
type sftpClientPool struct {
	available chan *client
	all       []*client
}

// newSFTPClientPool opens `size` independent SFTP sessions against the
// same server. Sessions are kept open until close() is called.
func newSFTPClientPool(ctx context.Context, req *core.DownloadRequest, size int) (*sftpClientPool, error) {
	if size < 1 {
		size = 1
	}
	all := make([]*client, 0, size)
	avail := make(chan *client, size)
	for i := 0; i < size; i++ {
		c, err := dial(ctx, req.URL, req.SSHKnownHosts, req.SSHInsecure)
		if err != nil {
			for _, existing := range all {
				existing.close()
			}
			return nil, err
		}
		all = append(all, c)
		avail <- c
	}
	return &sftpClientPool{available: avail, all: all}, nil
}

func (p *sftpClientPool) acquire() *client {
	return <-p.available
}

func (p *sftpClientPool) release(c *client) {
	p.available <- c
}

func (p *sftpClientPool) size() int {
	return len(p.all)
}

func (p *sftpClientPool) close() {
	for _, c := range p.all {
		c.close()
	}
}

// downloadRecursiveSequential downloads a remote directory using a
// single SFTP connection. Files are processed one at a time, with
// subdirectory recursion handled inline. This is the original
// behaviour; the parallel implementation lives in
// downloadRecursiveParallel and is enabled when req.RecursiveParallel
// is greater than one.
func downloadRecursiveSequential(ctx context.Context, req *core.DownloadRequest) (*core.DownloadResult, error) {
	c, err := dial(ctx, req.URL, req.SSHKnownHosts, req.SSHInsecure)
	if err != nil {
		return nil, err
	}
	defer c.close()

	return sftpDownloadDirStandalone(ctx, c, req)
}

// downloadRecursiveParallel downloads a remote directory using a pool
// of N SFTP connections. All workers share a single semaphore so the
// total in-flight goroutines never exceed N, regardless of directory
// depth.
func downloadRecursiveParallel(ctx context.Context, req *core.DownloadRequest) (*core.DownloadResult, error) {
	parallel := req.RecursiveParallel
	if parallel < 1 {
		parallel = 1
	}
	pool, err := newSFTPClientPool(ctx, req, parallel)
	if err != nil {
		return nil, err
	}
	defer pool.close()

	var (
		statsMu     sync.Mutex
		totalBytes  int64
		totalChunks int
		firstErr    error
	)

	sem := make(chan struct{}, parallel)
	var wg sync.WaitGroup

	setErr := func(err error) {
		statsMu.Lock()
		if firstErr == nil {
			firstErr = err
		}
		statsMu.Unlock()
	}

	wg.Add(1)
	sem <- struct{}{}
	go func() {
		defer wg.Done()
		defer func() { <-sem }()
		if ctx.Err() != nil {
			return
		}
		c := pool.acquire()
		defer pool.release(c)
		sftpDownloadDirWorker(ctx, c, req, &wg, sem, setErr, &statsMu, &totalBytes, &totalChunks, pool)
	}()

	wg.Wait()
	if firstErr != nil {
		return nil, firstErr
	}
	return &core.DownloadResult{
		BytesDownloaded: totalBytes,
		Duration:        0,
		Protocol:        "sftp",
		ChunksCount:     totalChunks,
	}, nil
}

// sftpDownloadDirStandalone is the original behaviour: walk the tree
// on a single client, download each file sequentially, aggregate
// stats inline.
func sftpDownloadDirStandalone(ctx context.Context, c *client, req *core.DownloadRequest) (*core.DownloadResult, error) {
	remotePath := req.URL.Path
	if remotePath == "" {
		remotePath = "."
	}

	absPath, err := c.realpath(remotePath)
	if err != nil {
		absPath = remotePath
	}

	outputDir := req.Output
	if outputDir == "" {
		outputDir = filepath.Base(absPath)
		if outputDir == "" || outputDir == "/" {
			outputDir = "download"
		}
	}

	startTime := time.Now()
	var totalBytes int64
	var totalChunks int

	if err := sftpWalkDir(ctx, c, absPath, outputDir, req, &totalBytes, &totalChunks); err != nil {
		return nil, err
	}

	return &core.DownloadResult{
		BytesDownloaded: totalBytes,
		Duration:        time.Since(startTime),
		Protocol:        "sftp",
		ChunksCount:     totalChunks,
	}, nil
}

// sftpDownloadDirWorker is the worker-pool variant. It lists the
// directory, then dispatches file downloads and subdirectory recursion
// through the shared semaphore. Subdirectory workers reuse the same
// pool to acquire clients. The return value is always nil; results
// and errors are routed through the shared fields.
func sftpDownloadDirWorker(
	ctx context.Context,
	c *client,
	req *core.DownloadRequest,
	wg *sync.WaitGroup,
	sem chan struct{},
	setErr func(error),
	statsMu *sync.Mutex,
	totalBytes *int64,
	totalChunks *int,
	pool *sftpClientPool,
) {
	remotePath := req.URL.Path
	if remotePath == "" {
		remotePath = "."
	}

	absPath, err := c.realpath(remotePath)
	if err != nil {
		absPath = remotePath
	}

	outputDir := req.Output
	if outputDir == "" {
		outputDir = filepath.Base(absPath)
		if outputDir == "" || outputDir == "/" {
			outputDir = "download"
		}
	}

	entries, err := sftpListDir(ctx, c, absPath)
	if err != nil {
		setErr(fmt.Errorf("sftp list %s: %w", absPath, err))
		return
	}

	if err := os.MkdirAll(outputDir, 0755); err != nil {
		setErr(fmt.Errorf("mkdir %s: %w", outputDir, err))
		return
	}

	for i, e := range entries {
		name := e.Name
		if name == "." || name == ".." {
			continue
		}
		isDir := e.Attr != nil && sftpIsDir(e.Attr.Mode)
		remoteChild := absPath + "/" + name
		localChild := filepath.Join(outputDir, name)
		_ = i

		if isDir {
			wg.Add(1)
			select {
			case sem <- struct{}{}:
				go func(remoteChild, localChild string) {
					defer wg.Done()
					defer func() { <-sem }()
					if ctx.Err() != nil {
						return
					}
					cc := pool.acquire()
					defer pool.release(cc)
					subReq := *req
					subReq.URL = &url.URL{Scheme: req.URL.Scheme, User: req.URL.User, Host: req.URL.Host, Path: remoteChild}
					subReq.Output = localChild
					sftpDownloadDirWorker(ctx, cc, &subReq, wg, sem, setErr, statsMu, totalBytes, totalChunks, pool)
				}(remoteChild, localChild)
			case <-ctx.Done():
				return
			}
		} else {
			wg.Add(1)
			select {
			case sem <- struct{}{}:
				go func(remoteChild, localChild string, size uint64) {
					defer wg.Done()
					defer func() { <-sem }()
					if ctx.Err() != nil {
						return
					}
					cc := pool.acquire()
					defer pool.release(cc)
					subReq := *req
					subReq.URL = &url.URL{Scheme: req.URL.Scheme, User: req.URL.User, Host: req.URL.Host, Path: remoteChild}
					subReq.Output = localChild
					subReq.Recursive = false
					subReq.RecursiveParallel = 0
					if err := sftpDownloadFile(ctx, cc, &subReq, size); err != nil {
						setErr(fmt.Errorf("sftp download %s: %w", remoteChild, err))
						return
					}
					statsMu.Lock()
					*totalBytes += int64(size)
					*totalChunks++
					statsMu.Unlock()
				}(remoteChild, localChild, e.Attr.Size)
			case <-ctx.Done():
				return
			}
		}
	}
}

// sftpWalkDir recursively walks one remote directory on `c`,
// downloading each file and recursing into subdirectories. Stats are
// accumulated inline. Used by the standalone mode.
func sftpWalkDir(ctx context.Context, c *client, remotePath, localPath string, req *core.DownloadRequest, totalBytes *int64, totalChunks *int) error {
	if ctx.Err() != nil {
		return ctx.Err()
	}

	if err := os.MkdirAll(localPath, 0755); err != nil {
		return fmt.Errorf("mkdir %s: %w", localPath, err)
	}

	entries, err := sftpListDir(ctx, c, remotePath)
	if err != nil {
		return fmt.Errorf("sftp list %s: %w", remotePath, err)
	}

	for _, e := range entries {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}

		name := e.Name
		if name == "." || name == ".." {
			continue
		}
		isDir := e.Attr != nil && sftpIsDir(e.Attr.Mode)
		remoteChild := remotePath + "/" + name
		localChild := filepath.Join(localPath, name)

		if isDir {
			if err := sftpWalkDir(ctx, c, remoteChild, localChild, req, totalBytes, totalChunks); err != nil {
				return err
			}
		} else {
			subReq := *req
			subReq.URL = &url.URL{Scheme: req.URL.Scheme, User: req.URL.User, Host: req.URL.Host, Path: remoteChild}
			subReq.Output = localChild
			subReq.Recursive = false
			subReq.RecursiveParallel = 0
			if err := sftpDownloadFile(ctx, c, &subReq, e.Attr.Size); err != nil {
				return fmt.Errorf("sftp download %s: %w", remoteChild, err)
			}
			*totalBytes += int64(e.Attr.Size)
			*totalChunks++
		}
	}
	return nil
}

// sftpListDir lists a single directory using the provided client and
// returns the entries. The client is not closed.
func sftpListDir(ctx context.Context, c *client, path string) ([]*dirEntry, error) {
	handle, err := c.opendir(path)
	if err != nil {
		return nil, err
	}
	defer c.closeHandle(handle)
	return c.readdir(handle)
}

// sftpDownloadFile downloads a single file from a pre-opened SFTP
// client. Unlike downloadSingleFile it does NOT honour the resume
// metadata and always treats each call as a fresh transfer — the
// caller is expected to set up req.Output correctly.
func sftpDownloadFile(ctx context.Context, c *client, req *core.DownloadRequest, sizeHint uint64) error {
	remotePath := req.URL.Path
	_ = sizeHint

	handle, err := c.open(remotePath)
	if err != nil {
		return fmt.Errorf("sftp open %s: %w", remotePath, err)
	}
	defer c.closeHandle(handle)

	if err := os.MkdirAll(filepath.Dir(req.Output), 0755); err != nil {
		return fmt.Errorf("mkdir %s: %w", filepath.Dir(req.Output), err)
	}
	f, err := os.OpenFile(req.Output, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
	if err != nil {
		return fmt.Errorf("open %s: %w", req.Output, err)
	}
	defer f.Close()

	offset := uint64(0)
	bufSize := c.maxPacket
	if bufSize > 32768 {
		bufSize = 32768
	}

	for {
		select {
		case <-ctx.Done():
			// Save partial progress so the user can resume. Best-effort:
			// failures here are non-fatal because the cancellation is
			// already in flight.
			saveSFTPResume(req, int64(offset))
			return ctx.Err()
		default:
		}

		data, err := c.read(handle, offset, bufSize)
		if err == io.EOF {
			// Successful transfer; drop any stale sidecar.
			_ = output.DeleteResumeMetadata(req.Output)
			return nil
		}
		if err != nil {
			return fmt.Errorf("sftp read: %w", err)
		}
		n, werr := f.Write(data)
		if werr != nil {
			return fmt.Errorf("write: %w", werr)
		}
		offset += uint64(n)
		if uint64(n) < uint64(len(data)) {
			_ = output.DeleteResumeMetadata(req.Output)
			return nil
		}
	}
}

// saveSFTPResume persists partial SFTP download progress to the
// standard `.goget.meta` sidecar. No-op for stdout writes or zero-byte
// transfers. Size hint is not used (SFTP parallel workers don't read
// attributes in the same loop) — we only persist what we actually know.
func saveSFTPResume(req *core.DownloadRequest, downloaded int64) {
	if req == nil || req.Output == "" || req.Output == "-" || downloaded <= 0 {
		return
	}
	meta := output.NewResumeMetadata(req.URL.String(), "", "", downloaded, 0)
	if err := meta.Save(req.Output); err != nil && req.Verbose {
		fmt.Fprintf(os.Stderr, "[sftp] Warning: failed to save resume metadata: %v\n", err)
	}
}