initialHEADmain

1 files changed, 186 insertions, 0 deletions

diff --git a/internal/cli/cmd_prune.go b/internal/cli/cmd_prune.go
new file mode 100644
index 0000000..14c3f0f
--- /dev/null
+++ b/internal/cli/cmd_prune.go
@@ -0,0 +1,186 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sync/atomic"
+	"time"
+
+	"github.com/spf13/pflag"
+
+	"tsne.dev/hopper/internal/log"
+	"tsne.dev/hopper/internal/source"
+)
+
+var _ command = (*pruneCommand)(nil)
+
+type pruneCommand struct {
+	flags *pflag.FlagSet
+	main  *mainCommand
+
+	concurrency uint
+	prefix      string
+	root        string
+	olderThan   duration
+}
+
+func newPruneCommand(main *mainCommand) *pruneCommand {
+	cmd := &pruneCommand{
+		flags: newFlagset("prune"),
+		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 consider live")
+	registerDurationFlag(cmd.flags, &cmd.olderThan, "older-than", "30d", "grace window, only orphans older than this are deleted")
+	registerDryRunFlag(cmd.flags)
+	return cmd
+}
+
+func (cmd *pruneCommand) Usage() string {
+	return "Usage:" +
+		"\n  hopper [global flags] prune [flags] ⟨source⟩" +
+		"\n" +
+		"\nFlags:" +
+		"\n" + cmd.flags.FlagUsages()
+}
+
+func (cmd *pruneCommand) Run(ctx context.Context, args []string) error {
+	parseFlags(cmd, cmd.flags, args)
+
+	if cmd.olderThan.isZero() {
+		return usageError{"grace window must not be zero"}
+	}
+
+	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"}
+	}
+
+	if err := cmd.main.lookupAccessKey(); err != nil {
+		return err
+	}
+
+	api := newStorageAPI(cmd.main.endpoint, cmd.main.zone, cmd.main.accessKey)
+	return cmd.prune(ctx, api, positionals[0], time.Now())
+}
+
+func (cmd *pruneCommand) prune(ctx context.Context, api storageAPI, sourceArg string, now time.Time) error {
+	start := time.Now()
+
+	live, err := cmd.liveSet(sourceArg)
+	if err != nil {
+		return err
+	}
+	if len(live) == 0 {
+		return errors.New("empty source (refusing to prune). check --root, --prefix, or the archive contents")
+	}
+
+	entries, err := listRemoteEntries(ctx, api, cmd.prefix)
+	if err != nil {
+		return err
+	}
+
+	cutoff := cmd.olderThan.cutoff(now)
+	candidates := make([]*remoteEntry, 0, len(entries)/4)
+	for i := range entries {
+		e := &entries[i]
+		if !e.isDir && !live[e.path] {
+			if e.lastChanged.IsZero() {
+				log.Warn("could not determine last modified date for %s (skipping)", e.path)
+			} else if e.lastChanged.Before(cutoff) {
+				candidates = append(candidates, e)
+			}
+		}
+	}
+
+	// Each delete is independent, so the pool never fails fast. A successful
+	// delete tombstones its entry.
+	pool := newWorkerPool(ctx, cmd.concurrency, false)
+	freed := int64(0)
+	for _, e := range candidates {
+		action := fmt.Sprintf("deleting %s (%s)", e.path, humanSize(e.size))
+		pool.submit(action, func(ctx context.Context) error {
+			if err := api.Delete(ctx, e.path); err != nil {
+				return fileError{path: e.path, err: err}
+			}
+			e.deleted = true
+			atomic.AddInt64(&freed, int64(e.size))
+			return nil
+		})
+	}
+
+	deleted, errs := pool.wait()
+
+	var cmdErr error
+	if reportFailures(ctx, "failed to delete files", errs) {
+		cmdErr = exitError{1}
+	}
+
+	cmd.removeEmptyDirs(ctx, api, entries)
+
+	failed := len(errs)
+	logSummary("deleted: %d"+
+		"\nfailed:  %d"+
+		"\nfreed:   %s"+
+		"\nelapsed: %s",
+		deleted, failed, humanSize64(freed), time.Since(start).Round(time.Millisecond),
+	)
+
+	return cmdErr
+}
+
+// removeEmptyDirs deletes directories left with no surviving children after the
+// file deletes and returns the number successfully removed. It runs as a
+// best-effort second pass: failures are reported but never propagated to the
+// exit code.
+func (cmd *pruneCommand) removeEmptyDirs(ctx context.Context, api storageAPI, entries []remoteEntry) {
+	dirs := emptyDirs(entries)
+	if len(dirs) == 0 {
+		return
+	}
+
+	pool := newWorkerPool(ctx, cmd.concurrency, false)
+	for _, dir := range dirs {
+		action := fmt.Sprintf("removing empty directory %s", dir.path)
+		pool.submit(action, func(ctx context.Context) error {
+			err := api.DeleteDir(ctx, dir.path)
+			if err != nil {
+				return fileError{path: dir.path, err: err}
+			}
+			return nil
+		})
+	}
+
+	_, errs := pool.wait()
+	if len(errs) > 0 {
+		_ = reportFailures(ctx, "failed to remove empty directories", errs)
+	}
+}
+
+// liveSet walks the source once and returns the set of prefixed storage keys in
+// the current release. It iterates entries only and never reads content.
+func (cmd *pruneCommand) liveSet(sourceArg string) (map[string]bool, error) {
+	src, err := source.NewTarGzSource(sourceArg, cmd.root)
+	if err != nil {
+		return nil, err
+	}
+	defer src.Close()
+
+	live := map[string]bool{}
+	for {
+		entry, err := src.Next()
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				return live, nil
+			}
+			return nil, err
+		}
+		live[prefixedPath(cmd.prefix, entry.Path)] = true
+	}
+}