package action

import (
	"context"
	"errors"
	"time"

	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	eventdata "github.com/tidbcloud/eventbus-datastruct/event"
	"github.com/tidbcloud/fastflow/pkg/entity/run"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/eventbus"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/regionalserver"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	exportrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/export"
	regionrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/displayevent"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/metric"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	PollExportAction = "poll_export"
)

type PollExportActionStruct struct {
	exportRepo        exportrepo.ExportRepo
	regionRepo        regionrepo.Repo
	regionalserverCli regionalserver.Client
	transaction       db.Transactioner
	eventCommitter    eventbus.WALCommitter
}

func NewPollExportAction(
	exportRepo exportrepo.ExportRepo,
	regionRepo regionrepo.Repo,
	regionalserverCli regionalserver.Client,
	transaction db.Transactioner,
	eventCommitter eventbus.WALCommitter,
) *PollExportActionStruct {
	return &PollExportActionStruct{
		exportRepo:        exportRepo,
		regionRepo:        regionRepo,
		regionalserverCli: regionalserverCli,
		transaction:       transaction,
		eventCommitter:    eventCommitter,
	}
}

func (a *PollExportActionStruct) Name() string {
	return PollExportAction
}

func (a *PollExportActionStruct) ParameterNew() any {
	return &common.ExportParameter{}
}

func (a *PollExportActionStruct) Run(ctx run.ExecuteContext, params any) error {
	cctx, span := workflow.GenerateContext(ctx)
	defer span.End()

	clusterId, exportId, err := common.GetExportParams(params)
	if err != nil {
		tilog.Error(cctx, "poll export action: invalid params", zap.Error(err))
		return err
	}

	tilog.Info(cctx, "poll export action start", zap.String("cluster_id", clusterId), zap.String("export_id", exportId))

	// poll interval is 10 seconds
	time.Sleep(10 * time.Second)

	ent, err := a.exportRepo.GetByClusterAndExportId(cctx, clusterId, exportId)
	if err != nil {
		tilog.Error(cctx, "create export action: failed to get export", zap.Error(err), zap.String("export_id", exportId))
		return err
	}

	region, err := a.regionRepo.Get(cctx, ent.RegionId)
	if err != nil {
		tilog.Error(cctx, "create export action: get region failed", zap.Error(err))
		return err
	}

	export, err := a.regionalserverCli.GetExport(cctx, region, &clusterngv1beta.GetExportRequest{
		ClusterId: clusterId,
		ExportId:  exportId,
	})
	if err != nil {
		tilog.Error(cctx, "get export from regional failed", zap.Error(err), zap.String("export_id", exportId))
		return err
	}

	tilog.Info(cctx, "get export from regional", zap.String("export_id", exportId), zap.String("state", export.GetState().String()))

	switch export.GetState() {
	case clusterngv1beta.Export_CREATING:
		recordExportRunningMetric(ent)
		return errors.New("export still creating, waiting for next poll")
	case clusterngv1beta.Export_RUNNING:
		if ent.SnapshotTime == nil {
			snapshotTime := export.GetSnapshotTime().AsTime()
			ent.SnapshotTime = &snapshotTime
			ent.SnapshotTso = export.GetSnapshotTso()
		}
		err = a.exportRepo.Save(cctx, ent)
		if err != nil {
			tilog.Error(cctx, "failed to save export", zap.Error(err))
		}
		recordExportRunningMetric(ent)
		return errors.New("export still creating, waiting for next poll")
	case clusterngv1beta.Export_SUCCEEDED:
		txErr := a.transaction.Transaction(cctx, func(ctx context.Context) error {
			latest, err := a.exportRepo.GetByClusterAndExportId(ctx, clusterId, exportId)
			if err != nil {
				return err
			}
			if latest.SnapshotTime == nil {
				snapshotTime := export.GetSnapshotTime().AsTime()
				latest.SnapshotTime = &snapshotTime
				latest.SnapshotTso = export.GetSnapshotTso()
			}
			completeTime := export.GetCompleteTime().AsTime()
			latest.CompleteTime = &completeTime
			eventAlreadyRecorded := latest.State == schema.ExportStateSucceeded
			latest.State = schema.ExportStateSucceeded
			if err := a.exportRepo.Save(ctx, latest); err != nil {
				tilog.Error(ctx, "failed to save export", zap.Error(err))
				return err
			}
			if !eventAlreadyRecorded {
				displayEv := newExportDisplayEvent(ctx, latest, displayevent.CreateExport, eventdata.EventStatusSucceeded, "Export succeeded.")
				displayevent.Write(ctx, a.eventCommitter, displayEv)
			}
			ent = latest
			return nil
		})
		if txErr != nil {
			return txErr
		}
		deleteExportRunningMetric(ent)
		recordExportTerminalStateMetric(ent)
		return nil
	case clusterngv1beta.Export_FAILED:
		txErr := a.transaction.Transaction(cctx, func(ctx context.Context) error {
			latest, err := a.exportRepo.GetByClusterAndExportId(ctx, clusterId, exportId)
			if err != nil {
				return err
			}
			if latest.SnapshotTime == nil {
				snapshotTime := export.GetSnapshotTime().AsTime()
				latest.SnapshotTime = &snapshotTime
				latest.SnapshotTso = export.GetSnapshotTso()
			}
			completeTime := export.GetCompleteTime().AsTime()
			latest.CompleteTime = &completeTime
			eventAlreadyRecorded := latest.State == schema.ExportStateFailed
			latest.State = schema.ExportStateFailed
			latest.Reason = export.GetReason()
			if err := a.exportRepo.Save(ctx, latest); err != nil {
				tilog.Error(ctx, "failed to save export", zap.Error(err))
				return err
			}
			if !eventAlreadyRecorded {
				displayEv := newExportDisplayEvent(ctx, latest, displayevent.CreateExport, eventdata.EventStatusFailed, "Export failed.")
				displayevent.Write(ctx, a.eventCommitter, displayEv)
			}
			ent = latest
			return nil
		})
		if txErr != nil {
			return txErr
		}
		deleteExportRunningMetric(ent)
		recordExportTerminalStateMetric(ent)
		return nil
	case clusterngv1beta.Export_CANCELED, clusterngv1beta.Export_DELETED:
		return nil
	default:
		tilog.Info(cctx, "export in unspecified state, retry", zap.String("export_id", exportId))
		return errors.New("export state unknown, waiting for next poll")
	}
}

func recordExportRunningMetric(ent *entity.Export) {
	if ent == nil || ent.CreatedAt.IsZero() {
		return
	}

	duration := time.Since(ent.CreatedAt).Seconds()
	if duration < 0 {
		return
	}
	metric.ExportRunningDurationSeconds.WithLabelValues(ent.ClusterId, ent.ExportId).Set(duration)
}

func deleteExportRunningMetric(ent *entity.Export) {
	if ent == nil {
		return
	}
	metric.ExportRunningDurationSeconds.DeleteLabelValues(ent.ClusterId, ent.ExportId)
}

func recordExportTerminalStateMetric(ent *entity.Export) {
	if ent == nil {
		return
	}

	metric.ExportTerminalStateTotal.WithLabelValues(ent.ClusterId, ent.ExportId, string(ent.State)).Inc()
}