initialHEADmain

1 files changed, 266 insertions, 0 deletions

diff --git a/internal/cli/cmd_push.go b/internal/cli/cmd_push.go
new file mode 100644
index 0000000..c58f75c
--- /dev/null
+++ b/internal/cli/cmd_push.go
@@ -0,0 +1,266 @@
+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])
+}