package tests

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path"
	"runtime"
	"sync"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/pingcap/dbaas/billing2/config"
	"github.com/pingcap/dbaas/billing2/db"
	"github.com/pingcap/dbaas/billing2/handler/eventbushandler"
	"github.com/pingcap/dbaas/billing2/handler/novuhandler"
	"github.com/pingcap/dbaas/billing2/handler/sqsmeasurehandler"
	"github.com/pingcap/dbaas/billing2/migrate"
	"github.com/pingcap/dbaas/billing2/utils/log"
	"github.com/pingcap/dbaas/billing2/utils/panicIf"
	"github.com/shopspring/decimal"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

var gormdb *gorm.DB = nil

var (
	dbOnce     sync.Once
	testDBName string
	tableNames []string
)

// ensureTestDB creates the test database and runs migrations once per process.
// The database and connection are reused across tests; each test only needs to clean data.
func ensureTestDB() {
	dbOnce.Do(func() {
		testDBName = fmt.Sprintf("unittest_dbaas_%v", uuid.New().ID())
		config.Get().Set(config.Database, testDBName)
		config.Get().Set(config.ReadOnlyMySQLDatabase, testDBName)
		log.Info("Creating test database (once)", zap.String("database", testDBName))

		var err error
		gormdb, err = db.GetSetupDB()
		panicIf.Error(err)
		setupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		setupDB := gormdb.WithContext(setupCtx)
		panicIf.Error(setupDB.Exec("SET SESSION lock_wait_timeout = 5").Error)
		panicIf.Error(setupDB.Exec(fmt.Sprintf("CREATE DATABASE %v CHARACTER SET utf8mb4 COLLATE utf8mb4_bin", testDBName)).Error)
		_, err = db.Initialize()
		panicIf.Error(err)
		panicIf.Error(migrate.MigrateDB(db.GetDB(), false))

		// Cache table names for fast cleanup
		panicIf.Error(gormdb.Raw(
			"SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = ? AND TABLE_TYPE = 'BASE TABLE'",
			testDBName,
		).Scan(&tableNames).Error)
		log.Info("Test database ready", zap.String("database", testDBName), zap.Int("tables", len(tableNames)))
	})
}

// cleanDB deletes all data from every table and resets auto-increment counters.
// Uses goroutine-per-table parallelism since FOREIGN_KEY_CHECKS=0 makes ordering irrelevant.
func cleanDB() {
	d := db.GetDB()
	cleanupConcurrency := config.Get().GetInt(config.MaxDatabaseConnections) / 2
	if cleanupConcurrency < 1 {
		cleanupConcurrency = 1
	}
	if cleanupConcurrency > 4 {
		cleanupConcurrency = 4
	}
	sem := make(chan struct{}, cleanupConcurrency)

	// Each goroutine gets its own session, so set FK checks per-session in each goroutine.
	// Keep cleanup concurrency bounded; package-level `go test ./...` already runs many
	// processes, and a goroutine per table can otherwise exhaust a shared local MySQL server.
	var wg sync.WaitGroup
	for _, table := range tableNames {
		wg.Add(1)
		go func(t string) {
			defer wg.Done()
			sem <- struct{}{}
			defer func() { <-sem }()

			sqlDB, err := d.DB()
			panicIf.Error(err)
			conn, err := sqlDB.Conn(context.Background())
			panicIf.Error(err)
			defer conn.Close()
			_, err = conn.ExecContext(context.Background(), "SET FOREIGN_KEY_CHECKS = 0")
			panicIf.Error(err)
			_, err = conn.ExecContext(context.Background(), fmt.Sprintf("DELETE FROM `%s`", t))
			panicIf.Error(err)
			_, err = conn.ExecContext(context.Background(), fmt.Sprintf("ALTER TABLE `%s` AUTO_INCREMENT = 1", t))
			panicIf.Error(err)
		}(table)
	}
	wg.Wait()
}

func setupTestConfigFromEnv() {
	result := map[string](interface{}){}
	mysqlTestConfig, ok := os.LookupEnv("MYSQL_TEST_CONFIG")
	if !ok {
		return
	}
	panicIf.Error(json.Unmarshal([]byte(mysqlTestConfig), &result))
	config.Get().Set(config.MYSQLPort, result["port"])
	config.Get().Set(config.MySQLHost, result["host"])
	config.Get().Set(config.MySQLUser, result["user"])
	config.Get().Set(config.MySQLPassword, result["password"])
	config.Get().Set(config.ReadOnlyMYSQLPort, result["port"])
	config.Get().Set(config.ReadOnlyMySQLHost, result["host"])
	config.Get().Set(config.ReadOnlyMySQLUser, result["user"])
	config.Get().Set(config.ReadOnlyMySQLPassword, result["password"])
}

type BillingTestSuite struct {
	suite.Suite
}

func (*BillingTestSuite) SetupTest() {
	Setup()
}

func (*BillingTestSuite) TearDownTest() {
	Shutdown()
}

func SetupConfig() {
	config.Get().Set(config.EnableIgnoreClusterInvalidatedMeasure, false)
	config.Get().Set(config.EnableRemoteConfig, false)
	config.Get().Set(config.IsRefuseDelayMeasure, false)
	config.Get().Set(config.BillsDisplayProjectAndClusterNameWithID, false)
	config.Get().Set(config.AttachProjectAndClusterNameWithIDOnMeasuring, false)
	_, codeFile, _, _ := runtime.Caller(0)
	config.Get().AddConfigPath(path.Join(path.Dir(codeFile), "../../config/"))
	panicIf.Error(config.Init())
	setupTestConfigFromEnv()
	config.Get().Set(config.MaxDatabaseConnections, 16)
	config.Get().Set(config.MaxDatabaseIDELConnections, 4)

	config.Get().Set(config.GlobalPricePackage, 0)
	// Inline shadow pricing is async (pool + reconcile timers) — keep it
	// off by default so existing suites stay deterministic; shadow-focused
	// suites opt in explicitly.
	config.Get().Set(config.EnablePricingInlineShadow, false)
}

func InitCache() {
	config.Get().Set(config.EnableCache, false)
}

func InitEventBus() {
	eb, err := NewLocalEventBusHandler()
	panicIf.Error(err)
	eventbushandler.Set(eb)
	panicIf.Error(eventbushandler.Get().Run())
}

func InitAuditLogSDK() {
	ctx := context.Background()
	panicIf.Error(auditlog.InitMockAuditLogSDK(ctx, time.Second))
}

func InitMeasureSender() {
	sqsmeasurehandler.SetMockMeasureSender()
}

func InitNovuHandler() {
	novuhandler.Set(LocalNovuHandlerImpl{})
}

func CloseEventbus() {
	panicIf.Error(eventbushandler.Get().Close())
}

func Setup() {
	SetupConfig()
	ensureTestDB()
	cleanDB()
	InitCache()
	InitEventBus()
	InitAuditLogSDK()
	InitMeasureSender()
	InitNovuHandler()
}

func Shutdown() {
	CloseEventbus()
}

type BillingAssert struct {
	*require.Assertions
	T *testing.T
}

func (assert *BillingAssert) Equal(expected interface{}, actual interface{}, msgAndArgs ...interface{}) {
	assert.T.Helper()
	expectedLeft, leftOk := expected.(decimal.Decimal)
	expectedRight, rightOk := actual.(decimal.Decimal)
	if leftOk && rightOk {
		if !expectedLeft.Equal(expectedRight) {
			assert.Fail(fmt.Sprintf("Not equal: \n"+
				"expected: %s\n"+
				"actual  : %s", expectedLeft.String(), expectedRight.String()), msgAndArgs...)
		}
		return
	}
	assert.Assertions.Equal(expected, actual, msgAndArgs...)
}

func (suite *BillingTestSuite) Assert() *BillingAssert {
	t := suite.T()
	rawAssert := require.New(t)
	return &BillingAssert{rawAssert, t}
}