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