vaultik/internal/vaultik/prune.go

package vaultik

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

	"sneak.berlin/go/vaultik/internal/log"
	"sneak.berlin/go/vaultik/internal/snapshot"
	"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")

// errMetadataDirDeleteFailed reports directories under metadata/ with no
// manifest that prune could not delete.
var errMetadataDirDeleteFailed = errors.New(
	"metadata director(ies) with no manifest 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 that has its
// manifest.
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"`
	// MetadataDirsDeleted counts the directories under metadata/ with no
	// manifest that prune deleted, and MetadataDirsFailed those it could
	// not delete.
	MetadataDirsDeleted int `json:"metadata_dirs_deleted"`
	MetadataDirsFailed  int `json:"metadata_dirs_failed,omitempty"`
}

// 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, and each directory
// under metadata/ that a backup left with no manifest.
func (v *Vaultik) PruneBlobs(opts *PruneOptions) error {
	log.Info("Starting prune operation")

	metadataObjects, err := v.listMetadataObjects()
	if err != nil {
		return err
	}

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

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

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

	dirsWithoutManifest, err := v.findMetadataDirsWithoutManifest(metadataObjects)
	if err != nil {
		return err
	}

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

	v.reportBlobsTooNew(tooNew, opts)

	if len(unreferencedBlobs) == 0 && len(dirsWithoutManifest) == 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),
		"metadata_dirs_without_manifest", len(dirsWithoutManifest))

	if !v.confirmPrune(opts, len(unreferencedBlobs), totalSize,
		dirsWithoutManifest) {
		return nil
	}

	v.deleteUnreferencedBlobs(unreferencedBlobs, allBlobs, result)
	v.deleteMetadataDirs(dirsWithoutManifest, result)

	if opts.JSON {
		err = v.outputPruneBlobsJSON(result)
		if err != nil {
			return err
		}
	} else {
		v.reportPruneDeletions(result, len(dirsWithoutManifest))
	}

	var failures []error

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

	if result.MetadataDirsFailed > 0 {
		failures = append(failures, fmt.Errorf("%d %w",
			result.MetadataDirsFailed, errMetadataDirDeleteFailed))
	}

	return errors.Join(failures...)
}

// confirmPrune states what prune found to delete, naming each directory
// under metadata/ by its remote key, and, unless --force or --json is
// given, asks the operator to confirm. It reports whether to go ahead.
func (v *Vaultik) confirmPrune(
	opts *PruneOptions, blobCount int, blobsSize int64, dirs []string,
) bool {
	if opts.JSON {
		return true
	}

	dirCount := len(dirs)
	blobsFound := fmt.Sprintf("Found %d unreferenced blob(s) totaling %s.",
		blobCount, ubytes(blobsSize))
	dirsFound := fmt.Sprintf("Found %d metadata director(ies) left by "+
		"backups that stopped before their manifest:", dirCount)

	if opts.Force {
		if blobCount > 0 {
			v.UI.Infof("%s", blobsFound)
		}

		if dirCount > 0 {
			v.UI.Infof("%s", dirsFound)

			for _, remoteKey := range dirs {
				v.UI.Detailf("metadata/%s/", remoteKey)
			}
		}

		return true
	}

	// The counts, the names, the prompt and its answer bypass the UI:
	// they are an interactive exchange the operator must see, so --quiet
	// does not silence them.
	if blobCount > 0 {
		v.stdoutf("%s\n", blobsFound)
	}

	if dirCount > 0 {
		v.stdoutf("%s\n", dirsFound)

		for _, remoteKey := range dirs {
			v.stdoutf("  metadata/%s/\n", remoteKey)
		}
	}

	switch {
	case dirCount == 0:
		v.stdoutf("\nDelete %d unreferenced blob(s)? [y/N] ", blobCount)
	case blobCount == 0:
		v.stdoutf("\nDelete %d metadata director(ies)? [y/N] ", dirCount)
	default:
		v.stdoutf("\nDelete %d unreferenced blob(s) and %d metadata "+
			"director(ies)? [y/N] ", blobCount, dirCount)
	}

	var confirm string

	// A failure to read the answer means no confirmation.
	_, err := v.scanStdin(&confirm)
	if err != nil || strings.ToLower(confirm) != "y" {
		v.printlnStdout("Cancelled")

		return false
	}

	return true
}

// reportPruneDeletions says how many blobs and metadata directories
// prune deleted, naming each kind only when it found some to delete.
func (v *Vaultik) reportPruneDeletions(result *PruneBlobsResult, dirCount int) {
	if result.BlobsFound > 0 {
		v.UI.Infof("Deleted %d blob(s) totaling %s.",
			result.BlobsDeleted, ubytes(result.BytesFreed))
	}

	if dirCount > 0 {
		v.UI.Infof("Deleted %d metadata director(ies) with no manifest.",
			result.MetadataDirsDeleted)
	}
}

// listMetadataObjects lists every object under metadata/ on the
// destination store.
func (v *Vaultik) listMetadataObjects() ([]storage.ObjectInfo, error) {
	log.Info("Listing remote snapshots")

	var objects []storage.ObjectInfo

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

		objects = append(objects, object)
	}

	return objects, nil
}

// collectReferencedBlobs downloads every manifest among the listed
// metadata objects 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(
	metadataObjects []storage.ObjectInfo,
) (map[string]bool, error) {
	// IDs returned by uniqueSnapshotIDs are remote keys (hashed
	// subdirectories under metadata/), not human snapshot IDs.
	remoteKeys := uniqueSnapshotIDs(metadataObjects)

	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
}

// uniqueSnapshotIDs returns deduplicated snapshot IDs from the listed
// metadata objects.
func uniqueSnapshotIDs(metadataObjects []storage.ObjectInfo) []string {
	seen := make(map[string]bool)

	var snapshotIDs []string

	for _, object := range metadataObjects {
		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
}

// 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
}

// findMetadataDirsWithoutManifest returns the remote key of each
// directory under metadata/ that has no manifest.json.zst, such as one a
// backup leaves when its manifest upload fails after db.zst.age was
// uploaded. It leaves out a directory a local snapshot record claims,
// and one holding an object newer than the prune grace period, or of
// unknown age because the listing gave no time for it: a backup still
// running may be writing there. A name that is not a remote key is
// never returned.
func (v *Vaultik) findMetadataDirsWithoutManifest(
	metadataObjects []storage.ObjectInfo,
) ([]string, error) {
	localIDs, err := v.localSnapshotIDs()
	if err != nil {
		return nil, fmt.Errorf("listing local snapshots: %w", err)
	}

	keep := make(map[string]bool, len(localIDs))
	for _, id := range localIDs {
		keep[snapshot.RemoteSnapshotKey(id)] = true
	}

	cutoff := time.Now().UTC().Add(-v.Config.PruneGracePeriod)
	listed := make(map[string]bool)

	for _, object := range metadataObjects {
		parts := strings.Split(object.Key, "/")
		if len(parts) < metadataKeyParts || !isBlobHash(parts[1]) {
			continue
		}

		remoteKey := parts[1]
		listed[remoteKey] = true

		if parts[2] == "manifest.json.zst" ||
			object.ModTime.IsZero() || object.ModTime.After(cutoff) {
			keep[remoteKey] = true
		}
	}

	var remoteKeys []string

	for remoteKey := range listed {
		if !keep[remoteKey] {
			remoteKeys = append(remoteKeys, remoteKey)
		}
	}

	sort.Strings(remoteKeys)

	return remoteKeys, nil
}

// deleteMetadataDirs deletes everything in the given directories under
// metadata/ and counts in the result the directories deleted and those
// it could not delete.
func (v *Vaultik) deleteMetadataDirs(
	remoteKeys []string, result *PruneBlobsResult,
) {
	for _, remoteKey := range remoteKeys {
		_, err := v.deleteRemoteSnapshotByKey(remoteKey)
		if err != nil {
			log.Error("Failed to delete metadata directory with no manifest",
				"remote_key", remoteKey, "error", err)

			result.MetadataDirsFailed++

			continue
		}

		log.Info("Deleted metadata directory with no manifest",
			"remote_key", remoteKey)

		result.MetadataDirsDeleted++
	}
}

// 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)
}