package tenants

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/tidbcloud/observability/vm-migration/internal/subprocess"
)

// S3ArchiveLister abstracts `aws s3api list-objects-v2 --prefix archived/<bizType>/`
// so tests inject fakes without spawning aws.
//
// Returns the list of cluster.yaml keys with their LastModified timestamps.
type S3ArchiveLister interface {
	List(ctx context.Context, bucket, bizType string) ([]ArchiveObject, error)
}

// S3ArchiveDownloader abstracts `aws s3 cp --recursive --include
// "*/cluster.yaml" s3://bucket/archived/<bizType>/ destDir/`.
//
// The destination is a local dir; the caller walks it after the bulk transfer
// completes.
type S3ArchiveDownloader interface {
	Download(ctx context.Context, bucket, bizType, destDir string) error
}

// ArchiveObject is one cluster.yaml entry as returned by the listing API.
type ArchiveObject struct {
	Key          string
	LastModified time.Time
}

// awsLister runs `aws s3api list-objects-v2`.
//
// Production-only; tests use fake listers.
type awsLister struct{}

// NewAWSLister wires a lister that shells out to the aws CLI (assumed on
// $PATH).
func NewAWSLister() S3ArchiveLister { return &awsLister{} }

func (l *awsLister) List(ctx context.Context, bucket, bizType string) ([]ArchiveObject, error) {
	prefix := fmt.Sprintf("%s/%s/", ArchivePrefix, bizType)
	raw, err := runCLI(ctx, AWSExecutable,
		[]string{"s3api", "list-objects-v2", "--bucket", bucket, "--prefix", prefix, "--output", "json"},
		fmt.Sprintf("aws s3api list-objects-v2 %s/%s", bucket, prefix))
	if err != nil {
		return nil, err
	}
	var doc struct {
		Contents []struct {
			Key          string `json:"Key"`
			LastModified string `json:"LastModified"`
		} `json:"Contents"`
	}
	if err := json.Unmarshal([]byte(raw), &doc); err != nil {
		return nil, fmt.Errorf("parse aws s3api output: %w", err)
	}
	out := make([]ArchiveObject, 0, len(doc.Contents))
	for _, obj := range doc.Contents {
		if !strings.HasSuffix(obj.Key, "/cluster.yaml") {
			continue
		}
		if obj.LastModified == "" {
			continue
		}
		t, err := parseISO8601(obj.LastModified)
		if err != nil {
			continue
		}
		out = append(out, ArchiveObject{Key: obj.Key, LastModified: t})
	}
	return out, nil
}

// awsDownloader runs `aws s3 cp --recursive --include "*/cluster.yaml" ...`,
// mirroring the per-file transfer lines to w.
type awsDownloader struct {
	w io.Writer
}

// NewAWSDownloader wires a downloader that shells out to the aws CLI and
// mirrors transfer output to w (pass the command's stderr).
func NewAWSDownloader(w io.Writer) S3ArchiveDownloader { return &awsDownloader{w: w} }

func (d *awsDownloader) Download(ctx context.Context, bucket, bizType, destDir string) error {
	uri := fmt.Sprintf("s3://%s/%s/%s/", bucket, ArchivePrefix, bizType)
	_, err := runCLIMirror(ctx, AWSExecutable,
		[]string{"s3", "cp", "--recursive", "--exclude", "*", "--include", "*/cluster.yaml", uri, destDir + "/"},
		fmt.Sprintf("aws s3 cp --recursive %s", uri),
		d.w)
	return err
}

// EntriesFromArchiveDir walks localDir recursively for cluster.yaml files,
// parses each, and returns matching Entries.
//
// Filters by bizTypes and labels every Entry with source. If validRelPaths is
// non-nil, only files whose path relative to localDir appears in the set are
// parsed — used by --max-age to filter the bulk download by listing-side
// timestamps.
func EntriesFromArchiveDir(localDir string, bizTypes []string, source string, validRelPaths map[string]bool) ([]Entry, error) {
	out := []Entry{}
	err := filepath.WalkDir(localDir, func(path string, d os.DirEntry, walkErr error) error {
		if walkErr != nil {
			return walkErr
		}
		if d.IsDir() {
			return nil
		}
		if filepath.Base(path) != "cluster.yaml" {
			return nil
		}
		rel, err := filepath.Rel(localDir, path)
		if err != nil {
			rel = path
		}
		rel = filepath.ToSlash(rel)
		if validRelPaths != nil && !validRelPaths[rel] {
			return nil
		}
		text, err := os.ReadFile(path)
		if err != nil {
			return err
		}
		entry, err := ParseClusterYAML(string(text), bizTypes, source)
		if err != nil {
			return err
		}
		if entry != nil {
			out = append(out, *entry)
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return out, nil
}

// relPathsSinceCutoff returns the set of relative paths (after stripping
// archived/<bizType>/) for objects whose LastModified is at or after cutoff.
//
// Objects whose key does not start with the bizType prefix are skipped (they
// belong to a different bizType's scan).
func relPathsSinceCutoff(bizType string, objects []ArchiveObject, cutoff time.Time) map[string]bool {
	prefix := fmt.Sprintf("%s/%s/", ArchivePrefix, bizType)
	out := map[string]bool{}
	for _, o := range objects {
		if o.LastModified.Before(cutoff) {
			continue
		}
		if rel, ok := strings.CutPrefix(o.Key, prefix); ok {
			out[rel] = true
		}
	}
	return out
}

// parseISO8601 parses an ISO 8601 timestamp from S3.
//
// Input format is e.g. '2024-01-15T12:34:56.000Z'. Returns the time in UTC.
func parseISO8601(text string) (time.Time, error) {
	for _, layout := range []string{
		time.RFC3339Nano,
		time.RFC3339,
		"2006-01-02T15:04:05.000Z",
		"2006-01-02T15:04:05Z",
	} {
		if t, err := time.Parse(layout, text); err == nil {
			return t.UTC(), nil
		}
	}
	return time.Time{}, fmt.Errorf("not ISO 8601: %q", text)
}

// firstNonEmpty returns the first argument whose trimmed form is non-empty.
func firstNonEmpty(values ...string) string {
	for _, v := range values {
		if s := strings.TrimSpace(v); s != "" {
			return s
		}
	}
	return ""
}

// runCLI runs a CLI subprocess (aws or kubectl), capturing stdout+stderr.
//
// On failure it returns an error whose message prefers stderr (trimmed) over
// the raw exec error, prefixed with desc. Centralizes the exec + firstNonEmpty
// error wrap shared by the aws/kubectl shell-outs in this package.
func runCLI(ctx context.Context, bin string, args []string, desc string) (string, error) {
	return runCLIMirror(ctx, bin, args, desc, nil)
}

// runCLIMirror is runCLI additionally mirroring stdout+stderr live to mirror
// (nil discards). Both streams are mirrored because aws s3 cp prints its
// per-file transfer lines on a version-dependent stream.
//
// PYTHONUNBUFFERED makes the aws CLI (Python) flush each transfer line
// immediately; without it stdout is block-buffered through the pipe and
// progress arrives in multi-minute bursts.
func runCLIMirror(ctx context.Context, bin string, args []string, desc string, mirror io.Writer) (string, error) {
	if mirror == nil {
		mirror = io.Discard
	}
	var stdout, stderr strings.Builder
	res := subprocess.Exec(subprocess.ExecOptions{
		Ctx:    ctx,
		Bin:    bin,
		Args:   args,
		Env:    append(os.Environ(), "PYTHONUNBUFFERED=1"),
		Stdout: io.MultiWriter(&stdout, mirror),
		Stderr: io.MultiWriter(&stderr, mirror),
	})
	if res.Err != nil {
		return "", fmt.Errorf("%s: %s", desc, firstNonEmpty(stderr.String(), res.Err.Error()))
	}
	return stdout.String(), nil
}