vaultik/internal/vaultik/prune.go

package vaultik

import (
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"sneak.berlin/go/vaultik/internal/log"
	"sneak.berlin/go/vaultik/internal/storage"
)

// PruneOptions contains options for the prune command
type PruneOptions struct {
	Force bool
	JSON  bool
}

// errNukeRequiresForce guards the destructive remote nuke operation.
var errNukeRequiresForce = errors.New(
	"nuke requires --force (this deletes ALL remote snapshots and blobs)")

// errBlobDeleteFailed reports unreferenced blobs prune could not delete.
var errBlobDeleteFailed = errors.New("blob(s) could not be deleted")

// errObjectsLeft reports objects remote nuke still finds on the backup
// destination store after deleting.
var errObjectsLeft = errors.New(
	"object(s) still on the backup destination store")

// nukeCommandHint is the command suggested when a remote nuke could not
// delete everything: running it again deletes what is left, which
// `vaultik prune` does not do for snapshot metadata.
const nukeCommandHint = "vaultik remote nuke --force"

// metadataDirName is the top-level remote directory holding snapshot
// metadata.
const metadataDirName = "metadata"

// blobKeyParts is the "/"-separated segment count of a blob object key
// (blobs/<aa>/<bb>/<hash>).
const blobKeyParts = 4

// NukeRemote deletes every object under metadata/ and blobs/ in remote
// storage, whatever its name, along with any object an upload cut off
// part-way left under a temporary `.partial` name, and last the
// destination marker. It returns successfully only when a listing
// afterwards finds nothing left under metadata/ and blobs/, and the next
// backup then starts from scratch.
//
// Refuses to run unless force is true. The caller is responsible for
// confirming with the user.
func (v *Vaultik) NukeRemote(force bool) error {
	if !force {
		return errNukeRequiresForce
	}

	// Checked before anything is deleted: running nuke again would fail
	// the same way, so the advice below must not follow these errors.
	// The listing also fails on a destination without the marker while
	// the local index holds snapshots.
	err := v.EnsureStorageBinding()
	if err != nil {
		return err
	}

	_, err = v.listAllRemoteSnapshotKeys()
	if err != nil {
		return err
	}

	err = v.emptyDestinationStore()
	if err != nil {
		v.UI.Warningf("Run '%s' again to delete what is left on the "+
			"backup destination store.", nukeCommandHint)

		return err
	}

	v.UI.Completef("Backup destination store is now empty.")

	return nil
}

// emptyDestinationStore does the deleting for NukeRemote and stops at
// the first step that fails. It also fails when objects are still
// listed at the end, naming each one.
func (v *Vaultik) emptyDestinationStore() error {
	v.UI.Beginf("Removing all snapshot metadata from backup destination store.")

	_, err := v.RemoveAllSnapshots(&RemoveOptions{Force: true})
	if err != nil {
		return fmt.Errorf("removing all snapshots: %w", err)
	}

	v.UI.Beginf("Removing everything still present in backup destination store.")

	var left []string

	for _, prefix := range []string{"metadata/", "blobs/"} {
		keys, err := v.deleteEverythingUnder(prefix)
		if err != nil {
			return err
		}

		left = append(left, keys...)
	}

	for _, key := range left {
		v.UI.Warningf("Still on the backup destination store: %s", key)
	}

	if len(left) > 0 {
		return fmt.Errorf("%d %w", len(left), errObjectsLeft)
	}

	// Last, once the local snapshot records are gone: the next backup
	// then has none and writes the marker again.
	err = v.Storage.Delete(v.ctx, destinationMarkerKey)
	if err != nil {
		return fmt.Errorf("deleting %s: %w", destinationMarkerKey, err)
	}

	return nil
}

// deleteEverythingUnder deletes every object under prefix, whatever its
// name, and returns the keys a listing afterwards still finds. A delete
// that fails is logged and the others still run; that object then
// shows up in the listing afterwards.
func (v *Vaultik) deleteEverythingUnder(prefix string) ([]string, error) {
	keys, err := v.Storage.List(v.ctx, prefix)
	if err != nil {
		return nil, fmt.Errorf("listing %s: %w", prefix, err)
	}

	for _, key := range keys {
		err = v.Storage.Delete(v.ctx, key)
		if err != nil {
			log.Error("Failed to delete object", "key", key, "error", err)
		}
	}

	// The file and rclone listings skip `.partial` objects, so the
	// deletes above never reach them.
	err = v.Storage.DeletePartialUploads(v.ctx, prefix)
	if err != nil {
		return nil, fmt.Errorf("deleting partial uploads: %w", err)
	}

	left, err := v.Storage.List(v.ctx, prefix)
	if err != nil {
		return nil, fmt.Errorf("listing %s: %w", prefix, err)
	}

	return left, nil
}

// PruneBlobsResult contains the result of a blob prune operation
//
//nolint:tagliatelle // snake_case is the established JSON output format
type PruneBlobsResult struct {
	BlobsFound   int   `json:"blobs_found"`
	BlobsDeleted int   `json:"blobs_deleted"`
	BlobsFailed  int   `json:"blobs_failed,omitempty"`
	BytesFreed   int64 `json:"bytes_freed"`
	// BlobsTooNew counts the unreferenced blobs kept because they are
	// newer than the prune grace period, or of unknown age. BlobsFound
	// does not include them.
	BlobsTooNew int `json:"blobs_too_new"`
}

// Prune removes orphaned data from the local index database AND
// unreferenced blobs from the backup destination store. This is the
// single user-facing prune entry point — the split between local and
// remote cleanup is an implementation detail. Calling code should
// prefer this method over PruneDatabase or PruneBlobs individually
// unless it specifically wants one half.
func (v *Vaultik) Prune(opts *PruneOptions) error {
	err := v.EnsureStorageBinding()
	if err != nil {
		return err
	}
	// First reconcile local snapshot records against remote metadata:
	// any local snapshot whose manifest is missing from the destination
	// store is treated as gone. This used to be the separate 'snapshot
	// cleanup' command and is now folded in so a single 'vaultik prune'
	// gets the local index fully back in sync with the destination.
	err = v.CleanupLocalSnapshots(opts)
	if err != nil {
		return fmt.Errorf("reconciling local snapshots with remote: %w", err)
	}

	_, err = v.PruneDatabase()
	if err != nil {
		return fmt.Errorf("pruning local database: %w", err)
	}

	return v.PruneBlobs(opts)
}

// PruneBlobs removes unreferenced blobs from storage
func (v *Vaultik) PruneBlobs(opts *PruneOptions) error {
	log.Info("Starting prune operation")

	allBlobsReferenced, err := v.collectReferencedBlobs()
	if err != nil {
		return err
	}

	allBlobs, err := v.listAllRemoteBlobs()
	if err != nil {
		return err
	}

	unreferencedBlobs, totalSize, tooNew := v.findUnreferencedBlobs(
		allBlobs, allBlobsReferenced)

	result := &PruneBlobsResult{
		BlobsFound:  len(unreferencedBlobs),
		BlobsTooNew: tooNew,
	}

	v.reportBlobsTooNew(tooNew, opts)

	if len(unreferencedBlobs) == 0 {
		log.Info("No unreferenced blobs found")

		if opts.JSON {
			return v.outputPruneBlobsJSON(result)
		}

		v.UI.Infof("No unreferenced blobs to remove.")

		return nil
	}

	log.Info("Found unreferenced blobs",
		"count", len(unreferencedBlobs), "total_size", ubytes(totalSize))

	if opts.Force && !opts.JSON {
		v.UI.Infof("Found %d unreferenced blob(s) totaling %s.",
			len(unreferencedBlobs), ubytes(totalSize))
	}

	if !opts.Force && !opts.JSON {
		// The count, the prompt and its answer bypass the UI: they are an
		// interactive exchange the operator must see, so --quiet does not
		// silence them.
		v.stdoutf("Found %d unreferenced blob(s) totaling %s.\n",
			len(unreferencedBlobs), ubytes(totalSize))
		v.stdoutf("\nDelete %d unreferenced blob(s)? [y/N] ", len(unreferencedBlobs))

		var confirm string

		_, err = v.scanStdin(&confirm)
		if err != nil {
			v.printlnStdout("Cancelled")

			return nil //nolint:nilerr // read failure means no confirmation
		}

		if strings.ToLower(confirm) != "y" {
			v.printlnStdout("Cancelled")

			return nil
		}
	}

	v.deleteUnreferencedBlobs(unreferencedBlobs, allBlobs, result)

	if opts.JSON {
		err = v.outputPruneBlobsJSON(result)
		if err != nil {
			return err
		}
	} else {
		v.UI.Infof("Deleted %d blob(s) totaling %s.",
			result.BlobsDeleted, ubytes(result.BytesFreed))
	}

	if result.BlobsFailed > 0 {
		return fmt.Errorf("%d %w", result.BlobsFailed, errBlobDeleteFailed)
	}

	return nil
}

// collectReferencedBlobs downloads all manifests and returns the set of
// referenced blob hashes.
//
// Every manifest must be read successfully. A manifest that cannot be
// downloaded or decoded means its snapshot's blobs are unknown, so
// treating them as unreferenced would let prune delete data a snapshot
// still needs. Rather than risk that silent loss, any failure returns an
// error naming the remote key and prune deletes nothing.
func (v *Vaultik) collectReferencedBlobs() (map[string]bool, error) {
	log.Info("Listing remote snapshots")
	// IDs returned by listUniqueSnapshotIDs are remote keys (hashed
	// subdirectories under metadata/), not human snapshot IDs.
	remoteKeys, err := v.listUniqueSnapshotIDs()
	if err != nil {
		return nil, fmt.Errorf("listing snapshot keys: %w", err)
	}

	log.Info("Found manifests in remote storage", "count", len(remoteKeys))

	allBlobsReferenced := make(map[string]bool)

	for _, remoteKey := range remoteKeys {
		log.Debug("Processing manifest", "remote_key", remoteKey)

		manifest, err := v.downloadManifestByKey(remoteKey)
		if err != nil {
			return nil, fmt.Errorf("reading manifest %s: %w", remoteKey, err)
		}

		for _, blob := range manifest.Blobs {
			allBlobsReferenced[blob.Hash] = true
		}
	}

	log.Info("Processed manifests",
		"count", len(remoteKeys), "unique_blobs_referenced", len(allBlobsReferenced))

	return allBlobsReferenced, nil
}

// listUniqueSnapshotIDs returns deduplicated snapshot IDs from remote metadata
func (v *Vaultik) listUniqueSnapshotIDs() ([]string, error) {
	objectCh := v.Storage.ListStream(v.ctx, "metadata/")
	seen := make(map[string]bool)

	var snapshotIDs []string

	for object := range objectCh {
		if object.Err != nil {
			return nil, fmt.Errorf("listing metadata objects: %w", object.Err)
		}

		parts := strings.Split(object.Key, "/")
		if len(parts) >= minSnapshotIDParts &&
			parts[0] == metadataDirName && parts[1] != "" {
			if strings.HasSuffix(object.Key, "/") ||
				strings.Contains(object.Key, "/manifest.json.zst") {
				snapshotID := parts[1]
				if !seen[snapshotID] {
					seen[snapshotID] = true
					snapshotIDs = append(snapshotIDs, snapshotID)
				}
			}
		}
	}

	return snapshotIDs, nil
}

// listAllRemoteBlobs returns a map of all blob hashes in remote storage
// to their listing entries.
func (v *Vaultik) listAllRemoteBlobs() (map[string]storage.ObjectInfo, error) {
	log.Info("Listing all blobs in storage")

	allBlobs := make(map[string]storage.ObjectInfo)
	blobObjectCh := v.Storage.ListStream(v.ctx, "blobs/")

	for object := range blobObjectCh {
		if object.Err != nil {
			return nil, fmt.Errorf("listing blobs: %w", object.Err)
		}

		parts := strings.Split(object.Key, "/")
		if len(parts) != blobKeyParts || parts[0] != "blobs" {
			continue
		}

		// The object name is read from the destination store and is not
		// trusted. A name that is not a blob hash (e.g. a short or
		// non-hex string) would panic the later hash[:2]/hash[2:4] path
		// build, so skip it with a warning rather than delete it.
		if !isBlobHash(parts[3]) {
			log.Warn("Skipping non-conforming object under blobs/",
				"key", object.Key)

			continue
		}

		allBlobs[parts[3]] = object
	}

	log.Info("Found blobs in storage", "count", len(allBlobs))

	return allBlobs, nil
}

// findUnreferencedBlobs returns the hashes of the blobs no manifest
// references that are older than the prune grace period, and their total
// size. It also returns how many unreferenced blobs it left out for
// being newer than that, or of unknown age because the listing gave no
// time for them.
func (v *Vaultik) findUnreferencedBlobs(
	allBlobs map[string]storage.ObjectInfo, referenced map[string]bool,
) ([]string, int64, int) {
	var (
		unreferenced []string
		totalSize    int64
		tooNew       int
	)

	cutoff := time.Now().UTC().Add(-v.Config.PruneGracePeriod)

	for hash, blob := range allBlobs {
		if referenced[hash] {
			continue
		}

		if blob.ModTime.IsZero() || blob.ModTime.After(cutoff) {
			tooNew++

			continue
		}

		unreferenced = append(unreferenced, hash)
		totalSize += blob.Size
	}

	return unreferenced, totalSize, tooNew
}

// reportBlobsTooNew says how many unreferenced blobs prune kept for being
// newer than the grace period or of unknown age. Under --json the
// document carries the count instead.
func (v *Vaultik) reportBlobsTooNew(tooNew int, opts *PruneOptions) {
	if tooNew == 0 || opts.JSON {
		return
	}

	v.UI.Infof("Kept %d unreferenced blob(s) newer than "+
		"prune_grace_period (%s), or of unknown age: a backup still "+
		"running may need them.",
		tooNew, v.Config.PruneGracePeriod)
}

// deleteUnreferencedBlobs deletes the given blobs from storage and
// populates the result.
func (v *Vaultik) deleteUnreferencedBlobs(
	unreferencedBlobs []string, allBlobs map[string]storage.ObjectInfo,
	result *PruneBlobsResult,
) {
	log.Info("Deleting unreferenced blobs")

	for i, hash := range unreferencedBlobs {
		blobPath := fmt.Sprintf("blobs/%s/%s/%s", hash[:2], hash[2:4], hash)

		err := v.Storage.Delete(v.ctx, blobPath)
		if err != nil {
			log.Error("Failed to delete blob", "hash", hash, "error", err)

			continue
		}

		result.BlobsDeleted++
		result.BytesFreed += allBlobs[hash].Size

		if (i+1)%progressLogEvery == 0 || i == len(unreferencedBlobs)-1 {
			log.Info("Deletion progress",
				"deleted", i+1,
				"total", len(unreferencedBlobs),
				"percent", fmt.Sprintf("%.1f%%",
					float64(i+1)/float64(len(unreferencedBlobs))*percentScale),
			)
		}
	}

	result.BlobsFailed = len(unreferencedBlobs) - result.BlobsDeleted

	log.Info("Prune complete",
		"deleted_count", result.BlobsDeleted,
		"deleted_size", ubytes(result.BytesFreed),
		"failed", result.BlobsFailed,
	)
}

// outputPruneBlobsJSON outputs the prune result as JSON
func (v *Vaultik) outputPruneBlobsJSON(result *PruneBlobsResult) error {
	encoder := json.NewEncoder(v.Stdout)
	encoder.SetIndent("", "  ")

	return encoder.Encode(result)
}