path: root/internal/cli/cmd_push.go

package cli

import (
	"context"
	"errors"
	"fmt"
	"io"
	"time"

	"github.com/spf13/pflag"

	"tsne.dev/hopper/internal/bunny"
	"tsne.dev/hopper/internal/log"
	"tsne.dev/hopper/internal/source"
)

// pruneHintFloor and pruneHintMultiple gate the post-push prune hint: the zone
// must hold at least floor files and more than multiple times the current
// release before stale assets are worth flagging. multiple covers the current
// release plus five retained rollbacks (six releases' worth of hashed assets).
const (
	pruneHintFloor    = 1000
	pruneHintMultiple = 6
)

var _ command = (*pushCommand)(nil)

type pushCommand struct {
	flags *pflag.FlagSet
	main  *mainCommand

	concurrency uint
	prefix      string
	root        string
}

func newPushCommand(main *mainCommand) *pushCommand {
	cmd := &pushCommand{
		flags: newFlagset("push"),
		main:  main,
	}
	cmd.flags.UintVar(&cmd.concurrency, "concurrency", 0, "number of concurrent transfers (0 = auto-detect based on CPUs)")
	registerPathPrefixFlag(cmd.flags, &cmd.prefix, "prefix", "", "path prefix within the zone to target")
	registerPathPrefixFlag(cmd.flags, &cmd.root, "root", "", "subtree of the source to upload")
	registerDryRunFlag(cmd.flags)
	return cmd
}

func (cmd *pushCommand) Usage() string {
	return "Usage:" +
		"\n  hopper [global flags] push [flags] ⟨source⟩" +
		"\n" +
		"\nFlags:" +
		"\n" + cmd.flags.FlagUsages()
}

func (cmd *pushCommand) Run(ctx context.Context, args []string) error {
	parseFlags(cmd, cmd.flags, args)

	positionals := cmd.flags.Args()
	if len(positionals) == 0 || positionals[0] == "" {
		return usageError{"missing source"}
	}
	if len(positionals) > 1 {
		return usageError{"only a single source is supported"}
	}
	source := positionals[0]

	if err := cmd.main.lookupAccessKey(); err != nil {
		return err
	}

	api := newStorageAPI(cmd.main.endpoint, cmd.main.zone, cmd.main.accessKey)
	return cmd.push(ctx, api, source)
}

func (cmd *pushCommand) push(ctx context.Context, api storageAPI, sourceArg string) error {
	src, err := source.NewTarGzSource(sourceArg, cmd.root)
	if err != nil {
		return err
	}
	defer src.Close()

	// Drain the complete remote listing before any writes. A List failure is
	// fatal here, so push never proceeds on a partial picture.
	// Index the remote listing by storage key so each source entry's skip /
	// conflict decision is an O(1) lookup rather than a linear scan per file
	// (the walk itself is O(remote)). A file and a directory can never share an
	// exact key in object storage, so keying by path is unambiguous.
	start := time.Now()

	remoteByPath := make(map[string]bunny.Object, 128)
	remoteFiles := 0
	it := api.List(ctx, cmd.prefix)
	for it.Next() {
		o := it.Object()
		if !o.IsDirectory {
			remoteFiles++
		}
		remoteByPath[o.Path] = o
	}
	if err := it.Err(); err != nil {
		return err
	}

	skipped := 0

	// stashedMutables holds mutable entries (e.g. *.html files) read during the
	// asset phase but not uploaded until all immutable assets are successfully
	// uploaded. The stash is in-memory. A release normally has only a handful of
	// mutable files only.
	stashedMutables := make([]mutableEntry, 0, 8)

	// Upload every immutable asset via the pool and stash mutables. The single
	// reader below buffers each entry's bytes and hands them to a fail-fast worker
	// pool. The first asset that fails to upload cancels the pool.
	assetsPool := newWorkerPool(ctx, cmd.concurrency, true)
	var readErr error
	for !assetsPool.cancelled() {
		entry, err := src.Next()
		if err != nil {
			if !errors.Is(err, io.EOF) {
				readErr = err
			}
			break
		}

		// Decide against the remote listing: a directory at this exact path is a
		// hard conflict; a matching-size file is skipped; any other case (absent,
		// size mismatch, 0/missing) falls through to re-upload.
		key := prefixedPath(cmd.prefix, entry.Path)
		remoteObj, remoteExists := remoteByPath[key]
		if remoteExists && remoteObj.IsDirectory {
			readErr = fmt.Errorf("remote path %q is a directory, release has a file there", key)
			break
		}

		if entry.IsMutable {
			data, err := entry.ReadContent(ctx)
			if err != nil {
				readErr = fmt.Errorf("read %q: %w", entry.Path, err)
				break
			}
			stashedMutables = append(stashedMutables, mutableEntry{path: key, data: data})
			continue
		}

		if remoteExists && remoteObj.Size == entry.Size {
			logAction("skipping %s (exists)", key)
			skipped++
			continue
		}

		data, err := entry.ReadContent(ctx)
		if err != nil {
			readErr = fmt.Errorf("read %q: %w", entry.Path, err)
			break
		}

		action := fmt.Sprintf("uploading %s (%s)", key, humanSize(len(data)))
		if !assetsPool.submit(action, uploadTask(api, key, data)) {
			break
		}
	}

	// A reader error is fatal and abandons all in-flight uploads.
	if readErr != nil {
		assetsPool.cancel(errors.New("upload cancelled"))
	}

	assetsUploaded, errs := assetsPool.wait()
	reported := reportFailures(ctx, "failed to upload files", errs)
	if readErr != nil {
		return readErr
	} else if reported {
		return exitError{1}
	}

	// Upload all stashed mutable files. This batch is neither atomic nor ordered.
	// A partial mutable-phase failure can leave an entry point live while a mutable
	// file it references is not yet uploaded. Mitigation: We report an error and
	// a re-run is idempotent.
	mutablesPool := newWorkerPool(ctx, cmd.concurrency, false)
	for _, m := range stashedMutables {
		action := fmt.Sprintf("uploading %s (%s)", m.path, humanSize(len(m.data)))
		if !mutablesPool.submit(action, uploadTask(api, m.path, m.data)) {
			break // the parent context was cancelled
		}
	}

	// Commit phase (HTML) is drain-and-report: every stashed mutable is attempted
	// and all failures are aggregated. Unlike the pre-commit asset phase, uploads
	// here genuinely happened, so we print a summary with counts even on partial
	// failure — but the failure list goes out first, keeping the summary last.
	mutablesUploaded, mutErrs := mutablesPool.wait()
	mutablesFailed := len(mutErrs)

	sourceFiles := assetsUploaded + mutablesUploaded + skipped + mutablesFailed
	uploaded := assetsUploaded + mutablesUploaded

	if sourceFiles == 0 {
		log.Warn("source contains no files to upload (pushed nothing). check --root, --prefix, or the archive contents")
	}

	var cmdErr error
	if reportFailures(ctx, "failed to upload files", mutErrs) {
		cmdErr = exitError{1}
	}

	summary := "uploaded: %d" +
		"\nskipped:  %d" +
		"\nremote:   %d" +
		"\nfailed:   %d" +
		"\nelapsed:  %s"
	if shouldHintPrune(sourceFiles, remoteFiles) {
		summary += "\n\nHint: There are many stale files on the zone. Consider running hopper prune."
	}
	logSummary(
		summary,
		uploaded, skipped, remoteFiles, mutablesFailed, time.Since(start).Round(time.Millisecond),
	)

	return cmdErr
}

func shouldHintPrune(sourceFiles, remoteFiles int) bool {
	return sourceFiles > 0 &&
		remoteFiles >= pruneHintFloor &&
		remoteFiles > pruneHintMultiple*sourceFiles
}

// uploadTask builds a pool task that PUTs data and, on failure, tags the error
// with the path so the report can name the file.
func uploadTask(api storageAPI, path string, data []byte) func(context.Context) error {
	return func(ctx context.Context) error {
		if err := api.Upload(ctx, path, data); err != nil {
			return fileError{path: path, err: err}
		}
		return nil
	}
}

// mutableEntry is a mutable file (current rule: *.html) read during the asset
// phase and held in memory until the commit point. data is owned by this
// struct.
type mutableEntry struct {
	path string
	data []byte
}

func humanSize(n int) string {
	return humanSize64(int64(n))
}

func humanSize64(n int64) string {
	const unit = 1024
	if n < unit {
		return fmt.Sprintf("%d B", n)
	}
	div, exp := int64(unit), 0
	for x := n / unit; x >= unit; x /= unit {
		div *= unit
		exp++
	}
	return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp])
}