package storage

import (
	"context"
	"crypto/rand"
	"encoding/binary"
	"encoding/hex"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/hashicorp/go-memdb"
	"github.com/pingcap/log"
	"github.com/tidbcloud/aws-shared-provider/pkg/auditlog/util"
	"github.com/tidbcloud/aws-shared-provider/pkg/metaserver/config"
	metametrics "github.com/tidbcloud/aws-shared-provider/pkg/metaserver/metrics"
	"github.com/tidbcloud/pkgv2/gormutil"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
	"gorm.io/gorm"
)

const (
	keyspaceMetaTableName = "keyspaceMeta"
	clusterMetaTableName  = "clusterMeta"

	etcdBatchSize = 1000
	// clusterIDKey is the etcd key that stores the tidb cluster id.
	clusterIDKey       = "/pd/cluster_id"
	pingTimeout        = 3 * time.Second
	clusterIDByteWidth = 8
)

const (
	essentialPlanName    = "essential"
	zoneRedundantEnabled = "yes"
)

type memDBStorage struct {
	etcdCli *clientv3.Client
	mysqlDB *gorm.DB

	initRevision       int64
	keyspaceMetaPrefix string
	keyspaceMemDB      *memdb.MemDB

	clusterMemDB    *memdb.MemDB
	kubeClusterName string

	clusterEventMu               sync.RWMutex
	clusterEventEpoch            string
	clusterEventRewindLowerBound uint64
	lastClusterEventID           uint64
	lastClusterEvents            map[string]ClusterEvent
	lastClusterEventPollSuccess  time.Time
	lastRetainedEventScanSuccess time.Time
	// Published windows and their Events slices are immutable after Store.
	clusterEventWindow atomic.Pointer[ClusterEventWindow]
}

func NewMemDBStorage(cfg *config.Config) (storage Storage, err error) {
	mysqlConfigured, err := validateMySQLConfig(cfg)
	if err != nil {
		return nil, err
	}

	var (
		etcdCli      *clientv3.Client
		mysqlDB      *gorm.DB
		clusterMemDB *memdb.MemDB
	)
	// make sure that resources are cleaned up if an error occurs during initialization
	defer func() {
		if err != nil {
			if etcdCli != nil {
				etcdCli.Close()
			}
			if mysqlDB != nil {
				if sqlDB, dbErr := mysqlDB.DB(); dbErr == nil {
					sqlDB.Close()
				}
			}
		}
	}()

	etcdCli, err = clientv3.New(clientv3.Config{
		Endpoints:   cfg.Endpoints,
		TLS:         cfg.TLSConfig,
		DialTimeout: 5 * time.Second,
		DialOptions: []grpc.DialOption{
			grpc.WithKeepaliveParams(keepalive.ClientParameters{
				Time:                10 * time.Second,
				Timeout:             2 * time.Second,
				PermitWithoutStream: true,
			}),
		},
	})
	if err != nil {
		return nil, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), pingTimeout)
	defer cancel()

	if _, err = etcdCli.Status(ctx, cfg.Endpoints[0]); err != nil {
		return nil, err
	}
	keyspaceMemDB, err := memdb.NewMemDB(keyspaceMetaTableSchema())
	if err != nil {
		return nil, err
	}

	if mysqlConfigured {
		mysqlDB, err = connectDB(cfg)
		if err != nil {
			return nil, err
		}
		clusterMemDB, err = memdb.NewMemDB(clusterMetaTableSchema())
		if err != nil {
			return nil, err
		}
	}

	clusterEventEpoch, err := newClusterEventEpoch()
	if err != nil {
		return nil, err
	}

	store := &memDBStorage{
		etcdCli:       etcdCli,
		mysqlDB:       mysqlDB,
		keyspaceMemDB: keyspaceMemDB,
		clusterMemDB:  clusterMemDB,

		kubeClusterName:   cfg.KubeClusterName,
		clusterEventEpoch: clusterEventEpoch,
		lastClusterEvents: make(map[string]ClusterEvent),
	}
	store.publishClusterEventWindow(nil)
	return store, nil
}

func validateMySQLConfig(cfg *config.Config) (bool, error) {
	mysqlFields := []struct {
		name  string
		value string
	}{
		{name: "cloud-provider-type", value: cfg.CloudProviderType},
		{name: "rds-secret-arn", value: cfg.RdsSecretArn},
		{name: "region", value: cfg.Region},
	}

	missingFields := make([]string, 0, len(mysqlFields))
	for _, field := range mysqlFields {
		if field.value == "" {
			missingFields = append(missingFields, field.name)
		}
	}
	if len(missingFields) == len(mysqlFields) {
		return false, nil
	}
	if len(missingFields) > 0 {
		return false, fmt.Errorf(
			"incomplete mysql configuration: missing %s",
			strings.Join(missingFields, ", "),
		)
	}
	if cfg.KubeClusterName == "" {
		return false, errors.New("kube-cluster-name is required when mysql is configured")
	}
	return true, nil
}

func newClusterEventEpoch() (string, error) {
	var epochBytes [16]byte
	if _, err := rand.Read(epochBytes[:]); err != nil {
		return "", fmt.Errorf("failed to generate cluster event epoch: %w", err)
	}
	return hex.EncodeToString(epochBytes[:]), nil
}

func nextClusterEventEpoch(previous string) (string, error) {
	for {
		epoch, err := newClusterEventEpoch()
		if err != nil {
			return "", err
		}
		if epoch != previous {
			return epoch, nil
		}
	}
}

func (m *memDBStorage) setKeyspaceMetaPrefix(ctx context.Context) error {
	start := time.Now()
	resp, err := m.etcdCli.Get(ctx, clusterIDKey)
	if err != nil {
		return fmt.Errorf("fail to get tidb cluster id from etcd: %v", err)
	}
	if len(resp.Kvs) == 0 {
		return fmt.Errorf("failed to find tidb cluster ID")
	}
	value := resp.Kvs[0].Value
	if len(value) != clusterIDByteWidth {
		return fmt.Errorf("falformed tidb cluster ID")
	}
	clusterID := binary.BigEndian.Uint64(value)
	m.keyspaceMetaPrefix = fmt.Sprintf("/pd/%d/keyspaces/meta/", clusterID)
	log.Info("[storage] keyspace meta prefix set",
		zap.String("prefix", m.keyspaceMetaPrefix),
		zap.Duration("duration", time.Since(start)),
	)
	return nil
}

func (m *memDBStorage) Start(ctx context.Context) error {
	log.Info("[storage] initializing mem db storage")

	var wg sync.WaitGroup
	// Use a channel to collect errors from goroutines.
	errs := make(chan error, 2)

	wg.Add(1)
	go func() {
		defer wg.Done()
		if err := m.setKeyspaceMetaPrefix(ctx); err != nil {
			errs <- err
			return
		}
		if err := m.fetchAllKeyspace(ctx); err != nil {
			errs <- err
		}
	}()

	if m.mysqlDB != nil && m.clusterMemDB != nil {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := m.reloadAllClusterState(ctx); err != nil {
				errs <- err
			}
		}()
	}

	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			return err // Return the first error encountered.
		}
	}

	go m.watchKeyspaceChanges(ctx)
	if m.mysqlDB != nil && m.clusterMemDB != nil {
		go m.watchClusterEvents(ctx)
	}
	return nil
}

func (m *memDBStorage) Close() error {
	etcdErr := m.etcdCli.Close()
	var dbErr error
	if m.mysqlDB != nil {
		sqlDB, err := m.mysqlDB.DB()
		if err != nil {
			if etcdErr != nil {
				return errors.Join(etcdErr, err)
			}
			return err
		}
		dbErr = sqlDB.Close()
	}
	return errors.Join(etcdErr, dbErr)
}

func connectDB(cfg *config.Config) (*gorm.DB, error) {
	metametrics.MySQLConfigured.Set(1)
	rdsConfig, err := util.GetRdsConfig(cfg.CloudProviderType, cfg.RdsSecretArn, cfg.Region)
	if err != nil {
		metametrics.MySQLLastOperationSuccess.Set(0)
		return nil, err
	}
	if err := rdsConfig.Validate(); err != nil {
		metametrics.MySQLLastOperationSuccess.Set(0)
		return nil, err
	}
	connCfg := rdsConfig.ToGormutilConfig()

	db, err := gormutil.Connect(*connCfg)
	if err != nil {
		metametrics.MySQLLastOperationSuccess.Set(0)
		return nil, err
	}
	if err := db.Use(&mysqlMetricsPlugin{}); err != nil {
		metametrics.MySQLLastOperationSuccess.Set(0)
		pluginErr := fmt.Errorf("register mysql metrics plugin: %w", err)
		sqlDB, dbErr := db.DB()
		if dbErr == nil {
			dbErr = sqlDB.Close()
		}
		return nil, errors.Join(pluginErr, dbErr)
	}
	metametrics.MySQLLastOperationSuccess.Set(1)
	return db, nil
}