package tenants

import (
	"context"
	"errors"
	"fmt"
	"io"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/observability/vm-migration/internal/jsonio"
	"github.com/tidbcloud/observability/vm-migration/internal/printer"
	"github.com/tidbcloud/observability/vm-migration/internal/status"
)

// EssentialBizType labels every entry produced by tenants essential.
const EssentialBizType = "essential"

// essentialS3BizType is the fixed bizType segment of the S3 archive path
// and the cluster.yaml filter for Essential-tier clusters.
const essentialS3BizType = "prod-devtier"

// DefaultEssentialTimeout is the --timeout default for the mgmt and rds
// stages.
const DefaultEssentialTimeout = 5 * time.Minute

// DefaultEssentialS3Timeout is the --timeout default for the s3 stage: it
// covers one aws s3 cp subprocess per deleted cluster, run sequentially.
const DefaultEssentialS3Timeout = 20 * time.Minute

// EssentialProviders are the accepted --provider values, as carried by mgmt
// ClusterFilter.CloudProviders and EssentialCluster.CloudProvider.
var EssentialProviders = []string{"aws", "gcp", "azure", "alicloud"}

// EssentialMgmtOptions carries inputs to the mgmt listing stage.
type EssentialMgmtOptions struct {
	Addr     string
	Region   string
	Provider string
	MaxAge   string
	Timeout  time.Duration
	Output   string
}

// EssentialRdsOptions carries inputs to the regional-server join stage.
type EssentialRdsOptions struct {
	ClustersIn string
	SecretFile string
	DBHost     string
	TLS        bool
	Timeout    time.Duration
	Output     string
}

// EssentialS3Options carries inputs to the S3 archive-resolution stage.
type EssentialS3Options struct {
	ClustersIn string
	S3Bucket   string
	Timeout    time.Duration
	Output     string
}

// Validate checks the required mgmt-stage inputs and parses MaxAge.
func (o EssentialMgmtOptions) Validate() (time.Duration, error) {
	if o.Addr == "" {
		return 0, errors.New("--addr is required (mgmt gRPC host:port)")
	}
	if o.Region == "" {
		return 0, errors.New("--region is required")
	}
	if o.Provider == "" {
		return 0, fmt.Errorf("--provider is required (one of %s)", strings.Join(EssentialProviders, ", "))
	}
	if !slices.Contains(EssentialProviders, o.Provider) {
		return 0, fmt.Errorf("invalid --provider %q: must be one of %s", o.Provider, strings.Join(EssentialProviders, ", "))
	}
	if o.Timeout <= 0 {
		return 0, fmt.Errorf("invalid timeout %s: must be positive", o.Timeout)
	}
	return ParseMaxAge(o.MaxAge)
}

// Validate checks the rds-stage inputs.
func (o EssentialRdsOptions) Validate() error {
	if o.ClustersIn == "" {
		return errors.New("--clusters-in is required (an essential mgmt artifact)")
	}
	if o.Timeout <= 0 {
		return fmt.Errorf("invalid timeout %s: must be positive", o.Timeout)
	}
	return nil
}

// Validate checks the s3-stage inputs.
func (o EssentialS3Options) Validate() error {
	if o.ClustersIn == "" {
		return errors.New("--clusters-in is required (an essential rds artifact)")
	}
	if o.S3Bucket == "" {
		return errors.New("--s3-bucket is required")
	}
	if o.Timeout <= 0 {
		return fmt.Errorf("invalid timeout %s: must be positive", o.Timeout)
	}
	return nil
}

// filterEssentialClusters keeps all alive clusters and only deleted ones
// deleted after cutoff.
//
// Deleted rows with an unparseable or missing DeleteTime are kept
// conservatively.
func filterEssentialClusters(rows []EssentialCluster, cutoff time.Time) []EssentialCluster {
	out := rows[:0]
	for _, r := range rows {
		if r.Deleted && r.DeleteTime != "" {
			if t, err := time.Parse(time.RFC3339, r.DeleteTime); err == nil && !t.UTC().After(cutoff) {
				continue
			}
		}
		out = append(out, r)
	}
	return out
}

// essentialSummary aggregates the cluster report artifact.
type essentialSummary struct {
	Total        int    `json:"total"`
	ActiveCount  int    `json:"active"`
	DeletedCount int    `json:"deleted"`
	DBFoundCount int    `json:"dbFound"`
	Region       string `json:"region"`
	Provider     string `json:"provider"`
	ServicePlan  string `json:"servicePlan"`
}

// essentialReport is the staged cluster report artifact: the kept clusters
// after the provider and liveness filters, enriched with regional-server vm
// info.
type essentialReport struct {
	GeneratedAt string             `json:"generatedAt"`
	Summary     essentialSummary   `json:"summary"`
	Clusters    []EssentialCluster `json:"clusters"`
}

// listEssentialClusters runs the mgmt stage: both deletion states, merged by
// clusterId, liveness-filtered by maxAge.
func listEssentialClusters(
	ctx context.Context,
	lister MgmtClusterLister,
	region, provider string,
	maxAge time.Duration,
	stderr io.Writer,
) ([]EssentialCluster, error) {
	printer.Printf(stderr, "listing clusters from mgmt (region=%s, provider=%s, plan=%s)...", region, provider, essentialPlan)
	active, err := lister.List(ctx, region, provider, new(false))
	if err != nil {
		return nil, err
	}
	deleted, err := lister.List(ctx, region, provider, new(true))
	if err != nil {
		return nil, err
	}
	merged := mergeByClusterID(active, deleted)
	printer.Printf(stderr, "  mgmt: %d cluster(s) (%d active, %d deleted)", len(merged), len(active), len(deleted))
	return filterEssentialClusters(merged, time.Now().Add(-maxAge)), nil
}

// writeEssentialReport writes report to path, or stdout when path is empty.
func writeEssentialReport(path string, report essentialReport, stdout io.Writer) error {
	if path == "" {
		payload, err := jsonio.MarshalIndent(report)
		if err != nil {
			return err
		}
		_, err = stdout.Write(payload)
		return err
	}
	if err := jsonio.WriteFile(path, report); err != nil {
		return fmt.Errorf("write %s: %w", path, err)
	}
	return nil
}

// readEssentialReport loads an essential artifact written by the mgmt or rds
// stage.
//
// Stages trust the artifact as-is: liveness and provider filtering happened
// at the mgmt stage and are not re-applied.
func readEssentialReport(path string) (essentialReport, error) {
	var report essentialReport
	if err := jsonio.ReadFile(path, &report); err != nil {
		return essentialReport{}, fmt.Errorf("%s: %w", path, err)
	}
	if report.Clusters == nil {
		return essentialReport{}, fmt.Errorf("%s: expected JSON object with a 'clusters' array", path)
	}
	return report, nil
}

// destName returns path, or "stdout" when empty, for log lines.
func destName(path string) string {
	if path == "" {
		return "stdout"
	}
	return path
}

// RunEssentialMgmt is the mgmt stage of essential collection: it lists
// Essential-plan clusters for the region/provider (both deletion states,
// merged by clusterId), applies the liveness window once, and writes the
// un-enriched cluster report that feeds the rds stage.
func RunEssentialMgmt(ctx context.Context, opts EssentialMgmtOptions, lister MgmtClusterLister, stdout, stderr io.Writer) error {
	maxAge, err := opts.Validate()
	if err != nil {
		return err
	}

	ctx, cancel := context.WithTimeout(ctx, opts.Timeout)
	defer cancel()

	kept, err := listEssentialClusters(ctx, lister, opts.Region, opts.Provider, maxAge, stderr)
	if err != nil {
		return err
	}
	report := buildEssentialReport(opts.Region, opts.Provider, kept, nil)
	if err := writeEssentialReport(opts.Output, report, stdout); err != nil {
		return err
	}
	printer.Printf(stderr, "wrote %d cluster(s) (active=%d deleted=%d) to %s",
		report.Summary.Total, report.Summary.ActiveCount, report.Summary.DeletedCount, destName(opts.Output))
	return nil
}

// RunEssentialRds is the regional-server stage: it loads the mgmt artifact,
// joins alive clusters against the regional MySQL by application_id to attach
// vm account/project ids, and writes the enriched cluster report that feeds
// the s3 stage.
func RunEssentialRds(ctx context.Context, opts EssentialRdsOptions, querier VMInfoQuerier, stdout, stderr io.Writer) error {
	if err := opts.Validate(); err != nil {
		return err
	}

	ctx, cancel := context.WithTimeout(ctx, opts.Timeout)
	defer cancel()

	report, err := readEssentialReport(opts.ClustersIn)
	if err != nil {
		return err
	}
	printer.Printf(stderr, "loaded %d cluster(s) from %s (provider=%s)", len(report.Clusters), opts.ClustersIn, report.Summary.Provider)

	appIDs := make([]string, 0, len(report.Clusters))
	for _, c := range report.Clusters {
		if c.Deleted {
			continue
		}
		appIDs = append(appIDs, essentialAppID(c))
	}
	lookup, err := querier.QueryVMInfo(ctx, appIDs)
	if err != nil {
		return err
	}

	enriched := buildEssentialReport(report.Summary.Region, report.Summary.Provider, report.Clusters, lookup)
	printer.Printf(stderr, "  rds: found %d/%d alive cluster(s)", enriched.Summary.DBFoundCount, len(appIDs))
	if err := writeEssentialReport(opts.Output, enriched, stdout); err != nil {
		return err
	}
	printer.Printf(stderr, "wrote %d cluster(s) (active=%d deleted=%d dbFound=%d) to %s",
		enriched.Summary.Total, enriched.Summary.ActiveCount, enriched.Summary.DeletedCount,
		enriched.Summary.DBFoundCount, destName(opts.Output))
	return nil
}

// RunEssentialS3 is the archive stage: it loads the rds artifact, resolves
// deleted clusters' tenants from archived cluster.yaml, merges them with the
// alive clusters' tenants from the enriched rows, and writes the complete
// Essential tenants artifact for the collected region/provider.
func RunEssentialS3(ctx context.Context, opts EssentialS3Options, fetcher S3Fetcher, stdout, stderr io.Writer) error {
	if err := opts.Validate(); err != nil {
		return err
	}

	ctx, cancel := context.WithTimeout(ctx, opts.Timeout)
	defer cancel()

	report, err := readEssentialReport(opts.ClustersIn)
	if err != nil {
		return err
	}
	printer.Printf(stderr, "loaded %d cluster(s) from %s", len(report.Clusters), opts.ClustersIn)

	entries, activeSkipped := entriesFromClusters(report.Clusters)
	mgmtCount := len(entries)
	deleted := []DeletedCluster{}
	for _, c := range report.Clusters {
		if c.Deleted {
			deleted = append(deleted, deletedClusterFromEssential(c))
		}
	}
	s3Entries, deletedSkipped := entriesFromS3Archive(ctx, fetcher, opts.S3Bucket, deleted, stderr)
	entries = append(entries, s3Entries...)

	s3Count := len(entries) - mgmtCount
	merged, err := writeEntries(opts.Output, entries, stdout)
	if err != nil {
		return err
	}
	s := report.Summary
	printer.Printf(stderr,
		"active=%d deleted=%d dbFound=%d dbMissing=%d entries=%d (mgmt=%d s3=%d) skipped=%d -> %s",
		s.ActiveCount, s.DeletedCount, s.DBFoundCount, s.ActiveCount-s.DBFoundCount,
		merged, mgmtCount, s3Count,
		activeSkipped+deletedSkipped, destName(opts.Output))
	return nil
}

// essentialAppID returns the regional-server application_id form of a
// cluster: its decimal ClusterID.
func essentialAppID(c EssentialCluster) string {
	return strconv.FormatUint(c.ClusterID, 10)
}

// deletedClusterFromEssential converts a report row to the exact-key archive
// lookup shape (string ids).
func deletedClusterFromEssential(c EssentialCluster) DeletedCluster {
	return DeletedCluster{
		ClusterID:      essentialAppID(c),
		ClusterName:    c.ClusterName,
		ProjectID:      strconv.FormatUint(c.ProjectID, 10),
		OrganizationID: strconv.FormatUint(c.OrganizationID, 10),
	}
}

// entriesFromS3Archive resolves deleted-cluster tenants from archived
// cluster.yaml via the shared fetchArchivedEntry unit.
//
// Returns the entries and the count of clusters that yielded none.
func entriesFromS3Archive(
	ctx context.Context,
	fetcher S3Fetcher,
	bucket string,
	clusters []DeletedCluster,
	stderr io.Writer,
) ([]Entry, int) {
	entries := []Entry{}
	skipped := 0
	for i, c := range clusters {
		key := clusterS3Key(essentialS3BizType, c)
		prefix := fmt.Sprintf("[%d/%d] s3://%s/%s name=%s", i+1, len(clusters), bucket, key, c.ClusterName)
		entry := fetchArchivedEntry(ctx, fetcher, bucket, essentialS3BizType, c, prefix, stderr)
		if entry == nil {
			skipped++
			continue
		}
		entries = append(entries, *entry)
	}
	return entries, skipped
}

// buildEssentialReport enriches kept clusters with the vm-info lookup and
// derives the summary counts in the same pass.
func buildEssentialReport(region, provider string, kept []EssentialCluster, lookup map[string]VMInfo) essentialReport {
	report := essentialReport{
		GeneratedAt: status.FormatTime(time.Now()),
		Summary:     essentialSummary{Region: region, Provider: provider, ServicePlan: essentialPlan.String()},
		Clusters:    kept,
	}
	for i := range report.Clusters {
		c := &report.Clusters[i]
		report.Summary.Total++
		if c.Deleted {
			report.Summary.DeletedCount++
		} else {
			report.Summary.ActiveCount++
		}
		d, ok := lookup[essentialAppID(*c)]
		if !ok {
			continue
		}
		c.DBFound = true
		report.Summary.DBFoundCount++
		c.InternalClusterID = d.InternalClusterID
		if d.VMAccountID.Valid {
			c.VMAccountID = d.VMAccountID.String
		}
		if d.VMProjectID.Valid {
			c.VMProjectID = d.VMProjectID.String
		}
		if d.DeletionTime.Valid {
			c.DBDeletionTime = status.FormatTime(d.DeletionTime.Time)
		}
	}
	return report
}

// entriesFromClusters builds tenant entries from enriched report rows.
//
// Deleted rows are not considered (they resolve via the S3 archive). An
// alive cluster yields an entry only when both vm ids are non-empty; there
// is no DefaultProjectID fallback. Returns the entries and the number of
// skipped alive clusters.
func entriesFromClusters(clusters []EssentialCluster) ([]Entry, int) {
	entries := []Entry{}
	skipped := 0
	for _, c := range clusters {
		if c.Deleted {
			continue
		}
		if !c.DBFound || c.VMAccountID == "" || c.VMProjectID == "" {
			skipped++
			continue
		}
		entries = append(entries, Entry{
			Tenant:  TenantID(c.VMAccountID, c.VMProjectID, ""),
			BizType: EssentialBizType,
			Source:  SourceMgmt,
		})
	}
	return entries, skipped
}