package tenants

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"strconv"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	commonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	tidbmgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"google.golang.org/grpc"
	"google.golang.org/protobuf/types/known/timestamppb"

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

type fakePageClient struct {
	pages   []*tidbmgmt.ListClustersResponse
	errAt   int
	err     error
	calls   int
	lastReq *tidbmgmt.ListClustersRequest
}

func (f *fakePageClient) ListClusters(ctx context.Context, req *tidbmgmt.ListClustersRequest, opts ...grpc.CallOption) (*tidbmgmt.ListClustersResponse, error) {
	defer func() { f.calls++ }()
	f.lastReq = req
	if f.errAt == f.calls && f.err != nil {
		return nil, f.err
	}
	if f.calls >= len(f.pages) {
		return &tidbmgmt.ListClustersResponse{}, nil
	}
	return f.pages[f.calls], nil
}

func fullPage(n int, idBase uint64) []*tidbmgmt.Cluster {
	clusters := make([]*tidbmgmt.Cluster, 0, n)
	for i := 0; i < n; i++ {
		clusters = append(clusters, &tidbmgmt.Cluster{ClusterId: idBase + uint64(i)})
	}
	return clusters
}

func TestListMgmtPagesShortPageStops(t *testing.T) {
	c := &fakePageClient{pages: []*tidbmgmt.ListClustersResponse{
		{Clusters: fullPage(essentialPageSize, 1), Total: essentialPageSize + 3},
		{Clusters: fullPage(3, essentialPageSize+1), Total: essentialPageSize + 3},
	}}
	rows, err := listMgmtPages(context.Background(), c, "ap-southeast-1", "aws", commonv1beta1.ServicePlan_Essential, nil)
	assert.NoError(t, err)
	assert.Len(t, rows, essentialPageSize+3)
	assert.Equal(t, 2, c.calls)
	assert.Equal(t, []string{"ap-southeast-1"}, c.lastReq.Filter.Regions)
	assert.Equal(t, []string{"aws"}, c.lastReq.Filter.CloudProviders)
	assert.Equal(t, commonv1beta1.ServicePlan_Essential, c.lastReq.Filter.ServicePlan)
	assert.Equal(t, uint32(essentialPageSize), c.lastReq.PageSize)
	assert.Equal(t, uint32(1), c.lastReq.Page)
}

func TestListMgmtPagesTotalStopsRunaway(t *testing.T) {
	// Server keeps returning full pages; Total guard stops the loop.
	c := &fakePageClient{pages: []*tidbmgmt.ListClustersResponse{
		{Clusters: fullPage(essentialPageSize, 1), Total: essentialPageSize},
		{Clusters: fullPage(essentialPageSize, 1), Total: essentialPageSize},
	}}
	rows, err := listMgmtPages(context.Background(), c, "r", "aws", commonv1beta1.ServicePlan_Essential, nil)
	assert.NoError(t, err)
	assert.Len(t, rows, essentialPageSize)
	assert.Equal(t, 1, c.calls)
}

func TestListMgmtPagesErrorWrapsPage(t *testing.T) {
	boom := errors.New("boom")
	c := &fakePageClient{
		pages: []*tidbmgmt.ListClustersResponse{
			{Clusters: fullPage(essentialPageSize, 1)},
			{Clusters: fullPage(essentialPageSize, 1)},
		},
		errAt: 2,
		err:   boom,
	}
	_, err := listMgmtPages(context.Background(), c, "r", "aws", commonv1beta1.ServicePlan_Essential, nil)
	assert.ErrorIs(t, err, boom)
	assert.Contains(t, err.Error(), "page 2")
}

func TestListMgmtPagesConvertsRows(t *testing.T) {
	delTS := time.Unix(1700000000, 0).UTC()
	createTS := time.Unix(1600000000, 0).UTC()
	c := &fakePageClient{pages: []*tidbmgmt.ListClustersResponse{
		{
			Clusters: []*tidbmgmt.Cluster{{
				ClusterId:      42,
				ClusterName:    "c1",
				ProjectId:      7,
				OrganizationId: 9,
				CloudProvider:  "aws",
				Region:         "ap-southeast-1",
				CreateTime:     timestamppb.New(createTS),
				DeleteTime:     timestamppb.New(delTS),
			}},
			Total: 1,
		},
	}}
	rows, err := listMgmtPages(context.Background(), c, "ap-southeast-1", "aws", commonv1beta1.ServicePlan_Essential, nil)
	assert.NoError(t, err)
	assert.Len(t, rows, 1)
	r := rows[0]
	assert.Equal(t, uint64(42), r.ClusterID)
	assert.Equal(t, "c1", r.ClusterName)
	assert.Equal(t, uint64(7), r.ProjectID)
	assert.Equal(t, uint64(9), r.OrganizationID)
	assert.Equal(t, "aws", r.CloudProvider)
	assert.Equal(t, "ap-southeast-1", r.Region)
	assert.True(t, r.Deleted)
	assert.NotEmpty(t, r.DeleteTime)
	assert.NotEmpty(t, r.CreateTime)
	assert.Empty(t, r.UpdateTime)
}

func TestMergeByClusterIDLaterListWins(t *testing.T) {
	active := []EssentialCluster{{ClusterID: 1, ClusterName: "active"}, {ClusterID: 2}}
	deleted := []EssentialCluster{{ClusterID: 1, ClusterName: "deleted", Deleted: true, DeleteTime: "2026-01-01T00:00:00Z"}}
	merged := mergeByClusterID(active, deleted)
	assert.Len(t, merged, 2)
	byID := map[uint64]EssentialCluster{}
	for _, c := range merged {
		byID[c.ClusterID] = c
	}
	assert.Equal(t, "deleted", byID[1].ClusterName)
	assert.True(t, byID[1].Deleted)
	assert.Equal(t, uint64(2), byID[2].ClusterID)
}

func TestMergeByClusterIDSortedOutput(t *testing.T) {
	active := []EssentialCluster{
		{ClusterID: 3},
		{ClusterID: 1},
		{ClusterID: 5},
	}
	deleted := []EssentialCluster{
		{ClusterID: 4, Deleted: true},
		{ClusterID: 2, Deleted: true},
	}
	merged := mergeByClusterID(active, deleted)
	ids := make([]uint64, 0, len(merged))
	for _, c := range merged {
		ids = append(ids, c.ClusterID)
	}
	assert.Equal(t, []uint64{1, 2, 3, 4, 5}, ids)
}

func TestFilterEssentialClusters(t *testing.T) {
	now := time.Now().UTC()
	cutoff := now.AddDate(0, 0, -90)
	rows := []EssentialCluster{
		{ClusterID: 1},
		{ClusterID: 2, Deleted: true, DeleteTime: now.AddDate(0, 0, -10).Format(time.RFC3339)},
		{ClusterID: 3, Deleted: true, DeleteTime: now.AddDate(0, 0, -200).Format(time.RFC3339)},
		{ClusterID: 4, Deleted: true, DeleteTime: ""},
		{ClusterID: 5, Deleted: true, DeleteTime: "not-a-time"},
	}
	kept := filterEssentialClusters(rows, cutoff)
	ids := []uint64{}
	for _, r := range kept {
		ids = append(ids, r.ClusterID)
	}
	assert.ElementsMatch(t, []uint64{1, 2, 4, 5}, ids)
}

func nullStr(s string) sql.NullString {
	if s == "" {
		return sql.NullString{}
	}
	return sql.NullString{String: s, Valid: true}
}

func nullTime(ts time.Time) sql.NullTime {
	return sql.NullTime{Time: ts, Valid: true}
}

func TestKeepBestVMInfo(t *testing.T) {
	t1 := time.Unix(1000, 0).UTC()
	t2 := time.Unix(2000, 0).UTC()

	// New row with vm info beats existing row without.
	best := map[string]VMInfo{"a": {VMAccountID: sql.NullString{}}}
	keepBestVMInfo(best, "a", VMInfo{VMAccountID: nullStr("1")})
	assert.Equal(t, "1", best["a"].VMAccountID.String)

	// Existing vm-populated row wins over non-populated newcomer.
	best = map[string]VMInfo{"a": {VMAccountID: nullStr("1")}}
	keepBestVMInfo(best, "a", VMInfo{VMAccountID: sql.NullString{}})
	assert.Equal(t, "1", best["a"].VMAccountID.String)

	// Both populated: latest update_timestamp wins; equal timestamps keep
	// the existing row.
	best = map[string]VMInfo{"a": {VMAccountID: nullStr("1"), UpdateTime: nullTime(t2)}}
	keepBestVMInfo(best, "a", VMInfo{VMAccountID: nullStr("2"), UpdateTime: nullTime(t1)})
	assert.Equal(t, "1", best["a"].VMAccountID.String)
	keepBestVMInfo(best, "a", VMInfo{VMAccountID: nullStr("3"), UpdateTime: nullTime(t2)})
	assert.Equal(t, "1", best["a"].VMAccountID.String)
	keepBestVMInfo(best, "a", VMInfo{VMAccountID: nullStr("3"), UpdateTime: nullTime(t2.Add(time.Second))})
	assert.Equal(t, "3", best["a"].VMAccountID.String)

	// Both without vm info: latest update_timestamp wins; null loses to valid.
	best = map[string]VMInfo{"a": {UpdateTime: sql.NullTime{}}}
	keepBestVMInfo(best, "a", VMInfo{UpdateTime: nullTime(t1)})
	assert.True(t, best["a"].UpdateTime.Valid)
}

func TestBatches(t *testing.T) {
	assert.Empty(t, batches(nil, 500))
	assert.Equal(t, [][]string{{"a"}}, batches([]string{"a"}, 500))
	big := make([]string, 1001)
	for i := range big {
		big[i] = strconv.Itoa(i)
	}
	got := batches(big, 500)
	assert.Len(t, got, 3)
	assert.Len(t, got[0], 500)
	assert.Len(t, got[1], 500)
	assert.Len(t, got[2], 1)
}

func TestResolveMySQLConfig(t *testing.T) {
	dir := t.TempDir()
	secret := filepath.Join(dir, "rds.json")
	assert.NoError(t, os.WriteFile(secret, []byte(`{"host":"rds.internal","port":3306,"username":"u","password":"p","dbname":"serverless"}`), 0o600))

	cfg, err := ResolveMySQLConfig(EssentialRdsOptions{SecretFile: secret})
	assert.NoError(t, err)
	assert.Equal(t, MySQLConfig{Host: "rds.internal", Port: 3306, User: "u", Password: "p", Name: "serverless"}, cfg)

	// --db-host overrides the secret host (port-forward case).
	cfg, err = ResolveMySQLConfig(EssentialRdsOptions{SecretFile: secret, DBHost: "127.0.0.1", TLS: true})
	assert.NoError(t, err)
	assert.Equal(t, MySQLConfig{Host: "127.0.0.1", Port: 3306, User: "u", Password: "p", Name: "serverless", TLS: true}, cfg)

	// Secret without a port falls back to defaultDBPort.
	noport := filepath.Join(dir, "noport.json")
	assert.NoError(t, os.WriteFile(noport, []byte(`{"host":"h","username":"u","password":"p","dbname":"n"}`), 0o600))
	cfg, err = ResolveMySQLConfig(EssentialRdsOptions{SecretFile: noport})
	assert.NoError(t, err)
	assert.Equal(t, defaultDBPort, cfg.Port)

	_, err = ResolveMySQLConfig(EssentialRdsOptions{})
	assert.ErrorContains(t, err, "--secret-file")
}

type fakeMgmtLister struct {
	active  []EssentialCluster
	deleted []EssentialCluster
	err     error
}

func (f *fakeMgmtLister) List(ctx context.Context, region, provider string, deleted *bool) ([]EssentialCluster, error) {
	if f.err != nil {
		return nil, f.err
	}
	if deleted != nil && *deleted {
		return f.deleted, nil
	}
	return f.active, nil
}

type fakeQuerier struct {
	lookup map[string]VMInfo
	err    error
}

func (f *fakeQuerier) QueryVMInfo(ctx context.Context, appIDs []string) (map[string]VMInfo, error) {
	if f.err != nil {
		return nil, f.err
	}
	return f.lookup, nil
}

func mgmtStageOpts() EssentialMgmtOptions {
	return EssentialMgmtOptions{
		Addr:     "a:1",
		Region:   "ap-southeast-1",
		Provider: "aws",
		MaxAge:   "90d",
		Timeout:  time.Minute,
	}
}

func TestEssentialMgmtValidate(t *testing.T) {
	maxAge, err := mgmtStageOpts().Validate()
	assert.NoError(t, err)
	assert.Equal(t, 90*24*time.Hour, maxAge)

	for _, tc := range []struct {
		name   string
		mutate func(*EssentialMgmtOptions)
		errMsg string
	}{
		{name: "addr", mutate: func(o *EssentialMgmtOptions) { o.Addr = "" }, errMsg: "--addr"},
		{name: "region", mutate: func(o *EssentialMgmtOptions) { o.Region = "" }, errMsg: "--region"},
		{name: "provider empty", mutate: func(o *EssentialMgmtOptions) { o.Provider = "" }, errMsg: "--provider is required"},
		{name: "provider invalid", mutate: func(o *EssentialMgmtOptions) { o.Provider = "digitalocean" }, errMsg: `invalid --provider "digitalocean"`},
		{name: "max-age", mutate: func(o *EssentialMgmtOptions) { o.MaxAge = "bogus" }, errMsg: "invalid duration"},
		{name: "timeout", mutate: func(o *EssentialMgmtOptions) { o.Timeout = 0 }, errMsg: "timeout"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			o := mgmtStageOpts()
			tc.mutate(&o)
			_, err := o.Validate()
			assert.ErrorContains(t, err, tc.errMsg)
		})
	}
}

func TestEssentialRdsAndS3Validate(t *testing.T) {
	rds := EssentialRdsOptions{ClustersIn: "in.json", SecretFile: "s.json", Timeout: time.Minute}
	assert.NoError(t, rds.Validate())
	rds.ClustersIn = ""
	assert.ErrorContains(t, rds.Validate(), "--clusters-in")
	rds.ClustersIn = "in.json"
	rds.Timeout = 0
	assert.ErrorContains(t, rds.Validate(), "timeout")

	s3 := EssentialS3Options{ClustersIn: "in.json", S3Bucket: "bkt", Timeout: time.Minute}
	assert.NoError(t, s3.Validate())
	s3.ClustersIn = ""
	assert.ErrorContains(t, s3.Validate(), "--clusters-in")
	s3.ClustersIn = "in.json"
	s3.S3Bucket = ""
	assert.ErrorContains(t, s3.Validate(), "--s3-bucket")
	s3.S3Bucket = "bkt"
	s3.Timeout = 0
	assert.ErrorContains(t, s3.Validate(), "timeout")
}

func TestEssentialStagedChain(t *testing.T) {
	dir := t.TempDir()
	mgmtOut := filepath.Join(dir, "essential-mgmt.json")
	rdsOut := filepath.Join(dir, "essential-rds.json")
	s3Out := filepath.Join(dir, "tenants-essential.json")

	// Stage 1: mgmt.
	lister := &fakeMgmtLister{
		active: []EssentialCluster{{ClusterID: 1, ClusterName: "a", ProjectID: 7, OrganizationID: 9}},
		deleted: []EssentialCluster{{
			ClusterID:      2,
			ClusterName:    "gone",
			ProjectID:      8,
			OrganizationID: 10,
			Deleted:        true,
			DeleteTime:     time.Now().AddDate(0, 0, -10).Format(time.RFC3339),
		}},
	}
	mgmtOpts := mgmtStageOpts()
	mgmtOpts.Output = mgmtOut
	var stdout, stderr bytes.Buffer
	assert.NoError(t, RunEssentialMgmt(context.Background(), mgmtOpts, lister, &stdout, &stderr))
	assert.Contains(t, stderr.String(), "wrote 2 cluster(s) (active=1 deleted=1)")

	// Stage 2: rds enriches alive rows only.
	querier := &fakeQuerier{lookup: map[string]VMInfo{
		"1": {InternalClusterID: "uuid-1", VMAccountID: nullStr("10"), VMProjectID: nullStr("20")},
	}}
	rdsOpts := EssentialRdsOptions{ClustersIn: mgmtOut, SecretFile: "unused", Timeout: time.Minute, Output: rdsOut}
	stdout.Reset()
	stderr.Reset()
	assert.NoError(t, RunEssentialRds(context.Background(), rdsOpts, querier, &stdout, &stderr))
	assert.Contains(t, stderr.String(), "loaded 2 cluster(s)")
	assert.Contains(t, stderr.String(), "  rds: found 1/1 alive cluster(s)")
	assert.Contains(t, stderr.String(), "wrote 2 cluster(s) (active=1 deleted=1 dbFound=1)")

	enriched, err := readEssentialReport(rdsOut)
	assert.NoError(t, err)
	for _, c := range enriched.Clusters {
		if c.Deleted {
			assert.False(t, c.DBFound)
		} else {
			assert.True(t, c.DBFound)
			assert.Equal(t, "10", c.VMAccountID)
			assert.Equal(t, "20", c.VMProjectID)
		}
	}

	// Stage 3: s3 resolves deleted rows and merges with alive entries.
	yaml := "bizType: prod-devtier\nvmAccountID: \"30\"\nvmProjectID: \"40\"\n"
	fetcher := &fakeS3Fetcher{bodies: map[string]string{
		"archived/prod-devtier/10/8/2/cluster.yaml": yaml,
	}}
	s3Opts := EssentialS3Options{ClustersIn: rdsOut, S3Bucket: "bucket", Timeout: time.Minute, Output: s3Out}
	stdout.Reset()
	stderr.Reset()
	assert.NoError(t, RunEssentialS3(context.Background(), s3Opts, fetcher, &stdout, &stderr))

	var doc entriesDoc
	assert.NoError(t, json.Unmarshal(mustReadFile(t, s3Out), &doc))
	assert.ElementsMatch(t, []Entry{
		{Tenant: "10:20", BizType: EssentialBizType, Source: SourceMgmt},
		{Tenant: "30:40", BizType: "prod-devtier", Source: SourceS3},
	}, doc.Tenants)
	assert.Contains(t, stderr.String(), "-> OK tenant=30:40")
	assert.Contains(t, stderr.String(), "entries=2 (mgmt=1 s3=1) skipped=0")
}

func TestRunEssentialRdsPropagatesErrors(t *testing.T) {
	boom := errors.New("boom")
	rdsOpts := EssentialRdsOptions{ClustersIn: "missing.json", SecretFile: "s", Timeout: time.Minute}
	var stdout, stderr bytes.Buffer
	err := RunEssentialRds(context.Background(), rdsOpts, &fakeQuerier{}, &stdout, &stderr)
	assert.ErrorContains(t, err, "missing.json")

	artifact := filepath.Join(t.TempDir(), "essential-mgmt.json")
	assert.NoError(t, jsonio.WriteFile(artifact, essentialReport{
		Summary:  essentialSummary{Total: 1, Region: "r", Provider: "aws"},
		Clusters: []EssentialCluster{{ClusterID: 1}},
	}))
	rdsOpts.ClustersIn = artifact
	err = RunEssentialRds(context.Background(), rdsOpts, &fakeQuerier{err: boom}, &stdout, &stderr)
	assert.ErrorIs(t, err, boom)
}

func TestRunEssentialS3StageFetchFailureSkips(t *testing.T) {
	artifact := filepath.Join(t.TempDir(), "essential-rds.json")
	assert.NoError(t, jsonio.WriteFile(artifact, essentialReport{
		Summary: essentialSummary{
			Total:        2,
			ActiveCount:  1,
			DeletedCount: 1,
			DBFoundCount: 1,
			Region:       "r",
			Provider:     "aws",
		},
		Clusters: []EssentialCluster{
			{ClusterID: 1, Deleted: false, DBFound: true, VMAccountID: "10", VMProjectID: "20"},
			{ClusterID: 2, Deleted: true, DeleteTime: time.Now().AddDate(0, 0, -10).Format(time.RFC3339)},
		},
	}))
	fetcher := &fakeS3Fetcher{errs: map[string]error{
		"archived/prod-devtier/0/0/2/cluster.yaml": errors.New("NotFound"),
	}}
	opts := EssentialS3Options{ClustersIn: artifact, S3Bucket: "bucket", Timeout: time.Minute}

	var stdout, stderr bytes.Buffer
	assert.NoError(t, RunEssentialS3(context.Background(), opts, fetcher, &stdout, &stderr))
	assert.Contains(t, stderr.String(), "-> SKIP (no cluster.yaml")
	assert.Contains(t, stderr.String(), "entries=1 (mgmt=1 s3=0) skipped=1")
}

func TestRunEssentialS3EmptyResultErrors(t *testing.T) {
	artifact := filepath.Join(t.TempDir(), "essential-rds.json")
	assert.NoError(t, jsonio.WriteFile(artifact, essentialReport{
		Summary:  essentialSummary{Region: "r", Provider: "aws"},
		Clusters: []EssentialCluster{},
	}))
	opts := EssentialS3Options{ClustersIn: artifact, S3Bucket: "bucket", Timeout: time.Minute}
	var stdout, stderr bytes.Buffer
	err := RunEssentialS3(context.Background(), opts, &fakeS3Fetcher{}, &stdout, &stderr)
	assert.ErrorContains(t, err, "no tenants found")
}

func TestRunEssentialS3StageSkipsMissingVMIDs(t *testing.T) {
	artifact := filepath.Join(t.TempDir(), "essential-rds.json")
	assert.NoError(t, jsonio.WriteFile(artifact, essentialReport{
		Summary: essentialSummary{
			Total:        3,
			ActiveCount:  2,
			DeletedCount: 1,
			DBFoundCount: 1,
			Region:       "r",
			Provider:     "aws",
		},
		Clusters: []EssentialCluster{
			{ClusterID: 1, DBFound: true, VMAccountID: "10", VMProjectID: "20"},
			{ClusterID: 2, DBFound: true, VMAccountID: "30"}, // missing project id
			{ClusterID: 3, Deleted: true, DeleteTime: time.Now().AddDate(0, 0, -10).Format(time.RFC3339)},
		},
	}))
	opts := EssentialS3Options{ClustersIn: artifact, S3Bucket: "bucket", Timeout: time.Minute}
	var stdout, stderr bytes.Buffer
	assert.NoError(t, RunEssentialS3(context.Background(), opts, &fakeS3Fetcher{}, &stdout, &stderr))
	assert.Contains(t, stderr.String(), "entries=1 (mgmt=1 s3=0) skipped=2")
}