package tenants

import (
	"fmt"
	"io"
	"sort"
	"strconv"
	"strings"

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

// Constants re-used by every tenant-list producer (tenants dedicated /
// essential) so the on-disk wire format stays consistent.
const (
	DefaultProjectID = "0"
	SourceK8s        = "k8s"
	SourceS3         = "s3"
	SourceMgmt       = "mgmt"

	// External CLI binaries. Resolved via $PATH at runtime; operators must
	// have valid credentials in their environment.
	KubectlExecutable = "kubectl"
	AWSExecutable     = "aws"

	ArchivePrefix = "archived"

	BizTypeLabel = "observability.pingcap.com/bizType"
)

// DedicatedBizTypes is the closed set of biz types considered "dedicated"
// for migration scoping.
var DedicatedBizTypes = []string{"tidbcloud", "staging-tidbcloud", "dev-tidbcloud"}

// sourcePriority ranks which source wins when the same tenant appears in
// multiple sources.
//
// Higher number wins. Archived sources outrank k8s because k8s no longer has
// those clusters once they're deleted. mgmt outranks both: the management
// service is authoritative for live Essential clusters, superseding even an
// archived cluster.yaml.
var sourcePriority = map[string]int{
	SourceK8s:  0,
	SourceS3:   1,
	SourceMgmt: 2,
}

// Entry is one tenant row in the output JSON.
type Entry struct {
	Tenant  string `json:"tenant"`
	BizType string `json:"bizType"`
	Source  string `json:"source"`
}

// ParseClusterYAML extracts bizType / vmAccountID / vmProjectID scalars from a
// cluster.yaml text, then returns an Entry only if bizType matches one of the
// requested values and a tenant ID can be formed.
//
// Returns nil (no error) when the yaml is valid but does not match the biz
// filter or lacks the tenant fields — callers treat this as "skip this
// cluster".
//
// The YAML parser is intentionally minimal: cluster.yaml is generated by the
// upstream control plane and contains only top-level scalar fields, so a full
// YAML parser would be over-engineering. Comments, blank lines, and quoted
// values are handled the same way as the v1 Python parser.
func ParseClusterYAML(text string, bizTypes []string, source string) (*Entry, error) {
	scalars := simpleYAMLScalars(text)
	matchedBiz := ""
	ok := false
	for _, b := range bizTypes {
		if scalars["biztype"] == b {
			matchedBiz = b
			ok = true
			break
		}
	}
	if !ok {
		return nil, nil
	}
	tenant := TenantID(scalars["vmaccountid"], scalars["vmprojectid"], DefaultProjectID)
	if tenant == "" {
		return nil, nil
	}
	return &Entry{Tenant: tenant, BizType: matchedBiz, Source: source}, nil
}

// TenantID combines account and project IDs into the canonical "acc:proj" form.
//
// Empty/whitespace-only account returns "". Empty project falls back to
// defaultProjectID. Whitespace is trimmed.
func TenantID(account, project, defaultProject string) string {
	account = strings.TrimSpace(account)
	if account == "" {
		return ""
	}
	project = strings.TrimSpace(project)
	if project == "" {
		project = defaultProject
	}
	return account + ":" + project
}

// simpleYAMLScalars extracts top-level "key: value" pairs from a yaml text.
//
// List items (lines starting with "-") are ignored because cluster.yaml in this
// codebase is flat.
func simpleYAMLScalars(text string) map[string]string {
	out := map[string]string{}
	for _, raw := range strings.Split(text, "\n") {
		line := strings.TrimSpace(raw)
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}
		if strings.HasPrefix(line, "-") {
			continue
		}
		idx := strings.Index(line, ":")
		if idx < 0 {
			continue
		}
		key := strings.TrimSpace(line[:idx])
		value := stripYAMLScalar(line[idx+1:])
		if value != "" {
			out[strings.ToLower(key)] = value
		}
	}
	return out
}

// stripYAMLScalar strips trailing comments and surrounding quotes from a scalar
// value.
func stripYAMLScalar(value string) string {
	value = strings.TrimSpace(value)
	if i := strings.Index(value, " #"); i >= 0 {
		value = strings.TrimRight(value[:i], " ")
	}
	if len(value) >= 2 && value[0] == value[len(value)-1] && (value[0] == '\'' || value[0] == '"') {
		return value[1 : len(value)-1]
	}
	return value
}

// mergeEntries deduplicates by Entry.Tenant.
//
// When the same tenant appears in multiple sources, the entry with the
// higher-priority source wins (higher sourcePriority value = archived source
// authoritative for deleted clusters). First-seen BizType is preserved on
// disagreement.
func mergeEntries(entries []Entry) []Entry {
	merged, _, _ := mergeEntriesTracked(entries)
	return merged
}

// mergeEntriesTracked is mergeEntries additionally reporting which input
// entry produced each merged row: origin[j] is the index into entries of the
// entry currently occupying merged slot j (replacements update it), and
// index maps tenant to slot.
//
// Used by tenants merge to attribute duplicates to their input files.
func mergeEntriesTracked(entries []Entry) ([]Entry, []int, map[string]int) {
	out := make([]Entry, 0, len(entries))
	origin := make([]int, 0, len(entries))
	index := map[string]int{}
	for i, e := range entries {
		j, ok := index[e.Tenant]
		if !ok {
			index[e.Tenant] = len(out)
			out = append(out, e)
			origin = append(origin, i)
			continue
		}
		if sourcePriority[e.Source] > sourcePriority[out[j].Source] {
			out[j] = e
			origin[j] = i
		}
	}
	return out, origin, index
}

// sortByTenant sorts entries in-place by Tenant in "acc:proj" natural order.
//
// Numeric by account first, then numeric by project. Non-numeric components
// sort after numeric. Stable so equal tenants keep insertion order from
// mergeEntries.
func sortByTenant(entries []Entry) {
	sort.SliceStable(entries, func(i, j int) bool {
		ki := tenantSortKey(entries[i].Tenant)
		kj := tenantSortKey(entries[j].Tenant)
		for k := 0; k < 2; k++ {
			if ki[k].less(kj[k]) {
				return true
			}
			if kj[k].less(ki[k]) {
				return false
			}
		}
		return false
	})
}

// tenantSortKey returns two comparable components for an "acc:proj" tenant id.
//
// Numeric components parse to int64; non-numeric ones fall through to a marker
// that sorts after every numeric value.
func tenantSortKey(tenant string) [2]keyPart {
	parts := strings.SplitN(tenant, ":", 2)
	for len(parts) < 2 {
		parts = append(parts, "")
	}
	return [2]keyPart{parseKeyPart(parts[0]), parseKeyPart(parts[1])}
}

// keyPart orders numeric values before non-numeric.
//
// Two numeric parts compare by int64 value; two non-numeric parts compare by
// string.
type keyPart struct {
	numeric bool
	num     int64
	str     string
}

func parseKeyPart(s string) keyPart {
	if n, err := strconv.ParseInt(s, 10, 64); err == nil {
		return keyPart{numeric: true, num: n, str: s}
	}
	return keyPart{numeric: false, str: s}
}

func (a keyPart) less(b keyPart) bool {
	if a.numeric && b.numeric {
		return a.num < b.num
	}
	if a.numeric != b.numeric {
		return a.numeric // numeric sorts before non-numeric
	}
	return a.str < b.str
}

// entriesDoc is the on-disk shape of tenants-*.json files.
//
// Used by both the writer (WriteEntriesFile) and the reader
// (ReadEntriesFile).
type entriesDoc struct {
	Total   int     `json:"total"`
	Tenants []Entry `json:"tenants"`
}

// ReadEntriesFile parses a tenants-*.json file in the canonical {"total": N,
// "tenants": [...]} shape written by WriteEntriesFile.
func ReadEntriesFile(path string) ([]Entry, error) {
	var doc entriesDoc
	if err := jsonio.ReadFile(path, &doc); err != nil {
		return nil, fmt.Errorf("%s: %w", path, err)
	}
	return doc.Tenants, nil
}

// WriteEntriesFile writes a tenants-*.json wire-shape document to path, or to
// stdout when path is empty.
//
// Used by the tenants producers so the on-disk format stays identical.
func WriteEntriesFile(path string, entries []Entry, stdout io.Writer) error {
	doc := entriesDoc{Total: len(entries), Tenants: entries}
	if path == "" {
		payload, err := jsonio.MarshalIndent(doc)
		if err != nil {
			return err
		}
		_, err = stdout.Write(payload)
		return err
	}
	return jsonio.WriteFile(path, doc)
}