package scheduler

import (
	"context"
	"encoding/base32"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/google/uuid"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/aws-shared-provider/pkg/auditlog/config"
	"github.com/tidbcloud/aws-shared-provider/pkg/auditlog/constant"
	"github.com/tidbcloud/aws-shared-provider/pkg/auditlog/metric"
	"github.com/tidbcloud/aws-shared-provider/pkg/auditlog/repo"
	"github.com/tidbcloud/aws-shared-provider/pkg/auditlog/util"
	"github.com/tidbcloud/aws-shared-provider/pkg/storage"
	"github.com/tidbcloud/aws-shared-provider/pkg/storage/migrate"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/pkgv2/db"

	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/serializer"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/tools/clientcmd"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/manager"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
)

type Scheduler struct {
	o11yS3Client         *s3.Client
	config               config.AuditLogConfig
	txProvider           db.TransProvider
	auditLogScheduleRepo repo.AuditLogScheduleRepo
	auditLogTaskRepo     repo.AuditLogTaskRepo
	auditLogWorker       *batchv1.Job
	manager              manager.Manager
}

func New(config config.AuditLogConfig) (*Scheduler, error) {
	if err := config.Check(); err != nil {
		return nil, err
	}

	o11yS3Clients, err := util.GetO11yS3Client(config.CloudProviderType, config.Region)
	if err != nil {
		return nil, fmt.Errorf("get o11y s3 clients err: %w", err)
	}

	rdsSecretRegion := config.Region
	if config.RdsSecretRegion != "" {
		rdsSecretRegion = config.RdsSecretRegion
	}
	rdsConfig, err := util.GetRdsConfig(config.CloudProviderType, config.RdsSecretArn, rdsSecretRegion)
	if err != nil {
		return nil, fmt.Errorf("get rds config err: %w", err)
	}

	db, err := storage.NewGormDB(rdsConfig)
	if err != nil {
		return nil, fmt.Errorf("new gorm db err: %w", err)
	}
	dbProvider, txProvider := storage.NewDBProvier(db)

	// migrate
	err = migrate.RunDBMigrate(db)
	if err != nil {
		return nil, fmt.Errorf("run db migrate err: %w", err)
	}

	// set up manager reconciler
	mgr, err := newManager()
	if err != nil {
		return nil, fmt.Errorf("new manager err: %w", err)
	}

	// audit log worker
	worker, err := decodeAuditLogWorker(context.TODO(), mgr, config.WorkerNs)
	if err != nil {
		return nil, fmt.Errorf("decode audit log worker err: %w", err)
	}

	// audit log task repo
	auditLogTaskRepo := repo.NewAuditLogTaskRepo(dbProvider)

	// metric
	alTaskcollector := metric.NewAuditLogTaskMetricCollector(auditLogTaskRepo)
	prometheus.MustRegister(alTaskcollector)

	reconciler := NewWorkerReconciler(mgr, auditLogTaskRepo)
	if err := reconciler.SetupWithManager(mgr); err != nil {
		return nil, fmt.Errorf("setup reconciler with manager err: %w", err)
	}

	return &Scheduler{
		o11yS3Client:         o11yS3Clients,
		config:               config,
		txProvider:           txProvider,
		auditLogScheduleRepo: repo.NewAuditLogScheduleRepo(dbProvider),
		auditLogTaskRepo:     auditLogTaskRepo,
		manager:              mgr,
		auditLogWorker:       worker,
	}, nil
}

func (s *Scheduler) StartManager() {
	if err := s.manager.Start(ctrl.SetupSignalHandler()); err != nil {
		panic(fmt.Sprintf("problem running manager %+v", err))
	}
}

func decodeAuditLogWorker(ctx context.Context, manager manager.Manager, workerNs string) (*batchv1.Job, error) {
	auditLogWorkerTemplateCm := &corev1.ConfigMap{}
	// call GetAPIReader to avoid cache is not initialized
	if err := manager.GetAPIReader().Get(ctx,
		client.ObjectKey{
			Name:      constant.AuditLogWorkerCmName,
			Namespace: workerNs,
		}, auditLogWorkerTemplateCm); err != nil {
		return nil, err
	}
	template := auditLogWorkerTemplateCm.Data[constant.AuditLogWorkerCmName]
	if template == "" {
		return nil, errors.New("template is empty")
	}

	auditLogWorker := &batchv1.Job{}
	codecs := serializer.NewCodecFactory(manager.GetScheme())
	if err := runtime.DecodeInto(codecs.UniversalDecoder(), []byte(template), auditLogWorker); err != nil {
		return nil, err
	}
	return auditLogWorker, nil
}

var base32Encoder = base32.StdEncoding.WithPadding(base32.NoPadding)

func GenerateIDWithPrefix(idPrefix string) string {
	uuidV4 := uuid.New()
	return idPrefix + strings.ToLower(base32Encoder.EncodeToString(uuidV4[:]))
}

func newManager() (manager.Manager, error) {
	scheme := runtime.NewScheme()
	utilruntime.Must(apiextensionsv1.AddToScheme(scheme))
	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
	utilruntime.Must(batchv1.AddToScheme(scheme))

	restConfig, err := clientcmd.BuildConfigFromFlags("", "")
	if err != nil {
		panic(fmt.Errorf("failed to load kubeconfig: %w", err))
	}

	labelSelector, err := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{
		MatchLabels: map[string]string{
			types.LabelJobType: constant.AuditLogWorkerLabelValue,
		},
	})
	if err != nil {
		return nil, fmt.Errorf("unable to create label selector: %w", err)
	}

	NoSyncPeriod := time.Duration(0)
	DefaultUnsafeDisableDeepCopy := true

	mgr, err := ctrl.NewManager(restConfig,
		ctrl.Options{
			Scheme:         scheme,
			Metrics:        metricsserver.Options{BindAddress: "0"},
			LeaderElection: false,
			Cache: cache.Options{
				SyncPeriod: &NoSyncPeriod,
				Scheme:     scheme,
				ByObject: map[client.Object]cache.ByObject{
					&batchv1.Job{}: {
						Label:                 labelSelector,
						UnsafeDisableDeepCopy: &DefaultUnsafeDisableDeepCopy,
					},
				},
			},
		},
	)
	if err != nil {
		return nil, fmt.Errorf("unable to create manager: %w", err)
	}
	return mgr, nil
}