package case_driver

import (
	"context"
	"fmt"
	"runtime/debug"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/dataflow-service-e2e/cases"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/config"
	"github.com/tidbcloud/dataflow-service-e2e/internal/log"
)

const (
	provisionTimeout = time.Minute * 40
)

type CaseDriver interface {
	SetUp()
	Run(t *testing.T)
	TearDown()
}

type casesDriver struct {
	lgr log.SimpleLogger

	// interruptCtx is cancelled on the first Ctrl-C to trigger a graceful
	// shutdown: running cases abort and suite teardown (cleanup) still runs.
	interruptCtx context.Context

	// clients
	dependenciesManager *dependenciesManager

	// provisionCancel releases the provisioning context created in SetUp.
	// It is held on the driver (instead of deferred) so the asynchronous
	// provisioning goroutines outlive SetUp and are released at TearDown.
	provisionCancel context.CancelFunc

	// internal stats
	// caseName: case
	cases2run map[string]cases.E2ECase
	asyncDeps bool
}

func NewCaseDriver(lgr log.SimpleLogger, interruptCtx context.Context, specifiedCases []string, specifiedCaseGroup string, asyncDeps bool) CaseDriver {
	cases := getCases2Run(lgr, specifiedCases, specifiedCaseGroup)
	return &casesDriver{
		lgr:          lgr,
		interruptCtx: interruptCtx,
		cases2run:    cases,
		asyncDeps:    asyncDeps,
		dependenciesManager: NewDependenciesManager(lgr, cases,
			NewTiDBDedicatedDependency(lgr), NewTiDBPremiumDependency(lgr), NewTiDBServerlessDependency(lgr),
			NewKafkaDependency(lgr), NewMySQLDependency(lgr), NewPulsarDependency(lgr),
			NewS3Dependency(lgr), NewGcsDependency(lgr), NewAzureBlobStorageDependency(lgr)),
	}
}

func (c *casesDriver) SetUp() {
	if c.interruptCtx != nil && c.interruptCtx.Err() != nil {
		// A signal received before setup should not start any new cloud
		// provisioning. Mark the barrier complete so the normal suite cleanup
		// path remains safe if it is registered by the caller.
		c.lgr.Warn("Skipping dependency provisioning because graceful interrupt was received before setup")
		c.provisionCancel = func() {}
		close(c.dependenciesManager.allProvisionDone)
		return
	}
	// Provisioning must outlive the case interrupt context. A graceful Ctrl-C
	// stops case execution, but a dependency create may already have created a
	// cloud resource before it reports that resource on its output channel.
	// Keep receiving until provisioning completes so teardown can clean the
	// complete inventory.
	ctx, cancel := context.WithTimeout(context.Background(), provisionTimeout)
	// The provisioning goroutines must outlive SetUp, so the cancel is held
	// on the driver and released during TearDown once provisioning is done.
	c.provisionCancel = cancel
	if c.asyncDeps {
		c.dependenciesManager.ProvisionAsync(ctx)
	} else {
		c.dependenciesManager.ProvisionAsync(ctx)
		<-c.dependenciesManager.allProvisionDone
	}
}

func (c *casesDriver) TearDown() {
	// Wait for async provisioning to finish before taking the final inventory
	// snapshot; otherwise a late entity could outlive the test process.
	c.lgr.Info("Teardown start: waiting for async provisioning to finish, asyncDeps: %t", c.asyncDeps)
	<-c.dependenciesManager.allProvisionDone
	failed := c.dependenciesManager.Cleanup()
	if c.provisionCancel != nil {
		c.provisionCancel()
	}
	if failed != nil {
		c.lgr.Error("Teardown failed: some driver-created dependencies could not be deleted; please remove them manually")
		panic("delete tidbClusters failed, please trouble shooting and remove them manually.")
	}
	c.lgr.Info("Teardown finished: driver-created dependencies cleaned up")
}

func (c *casesDriver) Run(t *testing.T) {
	for caseName, theCase := range c.cases2run {
		t.Run(caseName, func(t *testing.T) {
			perCaseLogger := log.NewCaseLogger(t, caseName)
			// Register cleanup before t.Parallel and dependency waiting so it
			// runs for success, FailNow, dependency failure, and recovered
			// panics. Cleanup errors are deliberately logged rather than
			// reported through testing.T so they do not change the case result.
			t.Cleanup(func() {
				if GlobalConfig == nil || GlobalConfig.MustCleanup == nil || !*GlobalConfig.MustCleanup {
					perCaseLogger.Info("case %q finished, per-case dependency release skipped (mustCleanup disabled)", caseName)
					return
				}
				perCaseLogger.Info("case %q finished, releasing per-case dependencies", caseName)
				c.dependenciesManager.ReleaseCase(caseName)
			})
			defer func() {
				if r := recover(); r != nil {
					perCaseLogger.Error("Stack trace of panic: %s", string(debug.Stack()))
					t.Fatalf("Run case '%s' panic, recover to make sure parent test can execute teardown, panic: %v", caseName, r)
				}
			}()
			t.Parallel()

			// Honor a graceful interruption (first Ctrl-C): the per-case
			// dependency wait and the case body abort when interruptCtx is
			// cancelled, letting the suite teardown run cleanup. Bind the
			// per-case logger to that context as well: generated API clients
			// pass it to the shared HTTP transport, which records response
			// status and x-debug-trace-id without serializing request bodies.
			interruptCtx := c.interruptCtx
			if interruptCtx == nil {
				interruptCtx = context.Background()
			}
			caseCtx := log.WithLogger(interruptCtx, perCaseLogger)
			theCase.SetContext(caseCtx)
			theCase.SetLogger(perCaseLogger)

			ctx, cancel := context.WithTimeout(caseCtx, time.Minute*40)
			defer cancel()
			if err := c.dependenciesManager.WaitForDeps(ctx, caseName); err != nil {
				if interruptCtx.Err() != nil {
					t.Logf("Case '%s' aborted due to graceful interrupt (Ctrl-C)", caseName)
					return
				}
				t.Fatalf("Case '%s' dependency provisioning failed: %v", caseName, err)
			}

			theCase.SetT(t)
			theCase.Run()
			t.Logf("Case '%s' is finished", caseName)
		})
	}
}

func getCases2Run(lgr log.SimpleLogger, specifiedCases []string, specifiedCaseGroup string) (cases2Run map[string]cases.E2ECase) {
	// caseGroup2Run accepts a comma-separated list of group names, so a single
	// run may combine multiple groups (e.g. "smoke,weeklyRegression").
	groupNames := splitCommaSeparated(specifiedCaseGroup)

	// Normalize the explicit case names first: blank entries (e.g. -cases=","
	// or a YAML list with whitespace) are dropped so an effectively empty
	// selection is rejected below instead of silently running nothing.
	caseNames := trimNonEmpty(specifiedCases)

	// validate
	if len(caseNames) == 0 && len(groupNames) == 0 {
		panic("both specifiedCases and specifiedCaseGroup are empty, no cases to run")
	}

	cases2Run = map[string]cases.E2ECase{}

	// handle specified cases; explicit cases and case groups combine as a
	// union, so both may be specified at the same time
	for _, name := range caseNames {
		theCase, ok := AllCases[name]
		if !ok {
			panic("case not found: " + name)
		}
		cases2Run[name] = theCase
	}

	// handle specified case groups
	for _, groupName := range groupNames {
		group, ok := CaseGroups[groupName]
		if !ok {
			panic("case group not found: " + groupName)
		}
		for _, theCase := range group {
			cases2Run[theCase.Name()] = theCase
		}
	}

	// A non-empty selection that resolves to zero cases (e.g. every selected
	// group was emptied by the domain filter) must fail, not report success.
	if len(cases2Run) == 0 {
		panic("no cases to run: the selection matched no cases for the effective domain")
	}

	sortedCasesNames := pie.Keys(cases2Run)
	sort.Strings(sortedCasesNames)
	fmtCases := make([]string, len(cases2Run))
	for k, v := range sortedCasesNames {
		fmtCases[k] = fmt.Sprintf("%d.%s", k+1, v)
	}
	lgr.Info("Cases to run:\n%s\n--------", strings.Join(fmtCases, "\n"))

	return cases2Run
}

// splitCommaSeparated splits a comma-separated list, trimming whitespace and
// dropping empty entries, so values like "smoke, weeklyRegression" or a
// trailing comma are accepted.
func splitCommaSeparated(s string) []string {
	items := []string{}
	for _, it := range strings.Split(s, ",") {
		if it = strings.TrimSpace(it); it != "" {
			items = append(items, it)
		}
	}
	return items
}

// trimNonEmpty trims whitespace from each item and drops blank entries, so a
// case list like ["", "  "] is treated as an empty selection.
func trimNonEmpty(items []string) []string {
	cleaned := []string{}
	for _, it := range items {
		if it = strings.TrimSpace(it); it != "" {
			cleaned = append(cleaned, it)
		}
	}
	return cleaned
}