package impl

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed"
	data_pipeline "github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline/planner"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	data_pipeline_repo "github.com/tidbcloud/dataflow-service-ng/app/repo/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service-ng/pkg/lake"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"go.uber.org/zap"
)

const defaultObservationPageSize = 100

var _ data_pipeline.ObservationModel = (*observationModel)(nil)

type ObservationConfig struct {
	PageSize       int
	StaleThreshold time.Duration
}

type observationModel struct {
	repo        data_pipeline_repo.DataPipelineRepo
	export      grpcli.ExportGrpcClient
	changefeeds biz_changefeed.BizChangefeedModel
	lake        lake.Client
	clock       planner.Clock
	config      ObservationConfig
}

func NewObservationModel(
	repo data_pipeline_repo.DataPipelineRepo,
	export grpcli.ExportGrpcClient,
	changefeeds biz_changefeed.BizChangefeedModel,
	lakeClient lake.Client,
	clock planner.Clock,
	config ObservationConfig,
) data_pipeline.ObservationModel {
	if config.PageSize <= 0 {
		config.PageSize = defaultObservationPageSize
	}
	return &observationModel{
		repo: repo, export: export, changefeeds: changefeeds,
		lake: lakeClient, clock: clock, config: config,
	}
}

func (m *observationModel) SyncObservations(ctx context.Context) bizerr.BizError {
	if m.repo == nil || m.clock == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "data pipeline observation dependencies are unavailable")
	}
	lastID := uint64(0)
	for {
		pipelines, err := m.repo.List(
			ctx,
			data_pipeline_repo.WithIDAfter(lastID),
			data_pipeline_repo.WithStatuses(
				entity.DataPipelineStatusCreating,
				entity.DataPipelineStatusRunning,
				entity.DataPipelineStatusPausing,
				entity.DataPipelineStatusPaused,
				entity.DataPipelineStatusResuming,
				entity.DataPipelineStatusFailed,
			),
			data_pipeline_repo.OrderByIDAsc(),
			data_pipeline_repo.WithPagination(0, m.config.PageSize),
		)
		if err != nil {
			return err.Chain("list data pipelines for observation")
		}
		if len(pipelines) == 0 {
			return nil
		}
		for index := range pipelines {
			pipeline := &pipelines[index]
			// One sweep visits many Pipelines, so the id is bound per Pipeline
			// rather than per sweep - otherwise everything the observation
			// writes would be attributed to whichever one came last.
			pipelineCtx := log.WithLogger(ctx, log.GetLogger(ctx).With(
				zap.Uint64("pipelineID", pipeline.ID.Uint64()),
			))
			if err := m.observeOne(pipelineCtx, pipeline); err != nil {
				log.GetLogger(pipelineCtx).Warn(
					"failed to persist data pipeline observation",
					zap.Error(err),
				)
			}
			lastID = pipeline.ID.Uint64()
		}
		if len(pipelines) < m.config.PageSize {
			return nil
		}
	}
}

func (m *observationModel) observeOne(
	ctx context.Context,
	pipeline *entity.DataPipeline,
) bizerr.BizError {
	now := m.clock.Now().UTC()
	observation := pipeline.Observation
	before := observation
	observation.LastAttemptAt = &now

	configuredChildren := 0
	successfulChildren := 0
	pollErrors := make([]string, 0, 3)
	lakeFullSyncCompleted := false
	if pipeline.Children.ExportID != "" {
		configuredChildren++
		if err := m.observeExport(ctx, pipeline, &observation); err != nil {
			pollErrors = append(pollErrors, "export")
			observation.ExportErrorMessage = "failed to observe Export"
		} else {
			successfulChildren++
		}
	}
	if pipeline.ExecutionPlan.Incremental != nil && pipeline.Children.ChangefeedID != 0 {
		configuredChildren++
		if err := m.observeChangefeed(ctx, pipeline, &observation); err != nil {
			pollErrors = append(pollErrors, "changefeed")
			observation.ChangefeedErrorMessage = "failed to observe Changefeed"
		} else {
			successfulChildren++
		}
	}
	destinationFailed := false
	if pipeline.Children.LakeTaskName != "" {
		configuredChildren++
		fullSyncCompleted, err := m.observeLake(ctx, pipeline, &observation, now)
		switch {
		case errors.Is(err, errTerminalDestinationRun):
			// Not a polling failure: the answer arrived and it is bad news.
			// Counting it as a successful observation keeps the row fresh
			// rather than stale, so the reason shown is the run failure
			// itself instead of "could not reach Lake".
			successfulChildren++
			destinationFailed = true
		case err != nil:
			pollErrors = append(pollErrors, "lake")
			observation.LakeErrorMessage = "failed to observe Lake integration"
		default:
			successfulChildren++
			lakeFullSyncCompleted = fullSyncCompleted
		}
	}

	expectedChildren := 2
	if pipeline.ExecutionPlan.Incremental != nil {
		expectedChildren++
	}
	observationSucceeded := configuredChildren == expectedChildren &&
		successfulChildren == expectedChildren && len(pollErrors) == 0
	if observationSucceeded {
		observation.LastSucceededAt = &now
		observation.PollError = nil
		observation.Stale = false
	} else if len(pollErrors) > 0 {
		observation.PollError = &entity.DataPipelineError{
			Code:       "OBSERVATION_FAILED",
			Message:    "failed to observe data pipeline child resources: " + strings.Join(pollErrors, ","),
			Retryable:  true,
			OccurredAt: &now,
		}
	}
	if observation.LastSucceededAt == nil ||
		(m.config.StaleThreshold > 0 && now.Sub(*observation.LastSucceededAt) > m.config.StaleThreshold) {
		observation.Stale = true
	}
	switch {
	case destinationFailed && observationSucceeded && isLiveStatus(pipeline.Status):
		// The destination has stopped and will not restart itself. Without
		// this the Pipeline sits in its last transitional state forever,
		// reporting a healthy observation of a task that is never going to
		// deliver another row.
		setPipelineStatus(ctx, pipeline, entity.DataPipelineStatusFailed, now)
		pipeline.LastError = &entity.DataPipelineError{
			Code:       lifecycleErrorChildFailed,
			Message:    observation.LakeErrorMessage,
			Retryable:  false,
			OccurredAt: &now,
		}
	case lakeFullSyncCompleted && observationSucceeded && !observation.Stale &&
		pipeline.Status == entity.DataPipelineStatusRunning &&
		pipeline.Params.SyncMode == entity.DataPipelineSyncModeFull &&
		pipeline.ExecutionPlan.Incremental == nil:
		setPipelineStatus(ctx, pipeline, entity.DataPipelineStatusCompleted, now)
	}
	observation.LatencySeconds = calculateLatencySeconds(
		observation.ChangefeedCheckpointTSO,
		observation.LakeResolvedTSO,
	)
	logObservationChange(ctx, pipeline, before, observation)
	pipeline.Observation = observation
	if err := m.repo.Save(ctx, pipeline); err != nil {
		return err.Chain("save data pipeline observation")
	}
	return nil
}

// logObservationChange records what a child's reported state became, and only
// when it changed.
//
// The loop runs every observation interval for every live Pipeline, so logging
// each pass would bury everything else. Logging only the transitions gives the
// history that matters - when a child started erroring, when it recovered, when
// the destination stopped - at a volume proportional to events rather than to
// time. Without it, a destination that reports a problem for twenty minutes
// produces no log line at all.
func logObservationChange(
	ctx context.Context,
	pipeline *entity.DataPipeline,
	before, after entity.DataPipelineObservation,
) {
	changed := make([]zap.Field, 0, 6)
	track := func(name string, was, now string) {
		if was != now {
			changed = append(changed, zap.String(name, was+" -> "+now))
		}
	}
	track("exportStatus", before.ExportStatus, after.ExportStatus)
	track("changefeedStatus", before.ChangefeedStatus, after.ChangefeedStatus)
	track("lakeDataSourceStatus", before.LakeDataSourceStatus, after.LakeDataSourceStatus)
	track("lakeIntegrationStatus", before.LakeIntegrationStatus, after.LakeIntegrationStatus)
	track("exportError", before.ExportErrorMessage, after.ExportErrorMessage)
	track("changefeedError", before.ChangefeedErrorMessage, after.ChangefeedErrorMessage)
	track("lakeError", before.LakeErrorMessage, after.LakeErrorMessage)
	if before.Stale != after.Stale {
		changed = append(changed, zap.Bool("stale", after.Stale))
	}
	if len(changed) == 0 {
		return
	}
	fields := append([]zap.Field{
		zap.Uint64("clusterID", pipeline.ClusterID.Uint64()),
		zap.String("pipelineStatus", string(pipeline.Status)),
	}, changed...)
	log.GetLogger(ctx).Info("data pipeline child state changed", fields...)
}

func (m *observationModel) observeExport(
	ctx context.Context,
	pipeline *entity.DataPipeline,
	observation *entity.DataPipelineObservation,
) error {
	if m.export == nil {
		return fmt.Errorf("Export observer is unavailable")
	}
	export, err := m.export.GetExport(ctx, exportCaller(pipeline), &globalngv1.GetExportRequest{
		ClusterId: strconv.FormatUint(pipeline.ClusterID.Uint64(), 10),
		ExportId:  pipeline.Children.ExportID,
	})
	if err != nil {
		return err
	}
	if export == nil {
		return fmt.Errorf("Export observation is empty")
	}
	observation.ExportStatus = export.GetState().String()
	observation.ExportErrorMessage = sanitizeChildMessage(
		export.GetReason(), pipeline.Params.Staging.S3.AccessKeyID)
	return nil
}

func (m *observationModel) observeChangefeed(
	ctx context.Context,
	pipeline *entity.DataPipeline,
	observation *entity.DataPipelineObservation,
) error {
	if m.changefeeds == nil {
		return fmt.Errorf("Changefeed observer is unavailable")
	}
	response, err := m.changefeeds.GetBizChangefeed(ctx, &domain.GetBizChangefeedReq{
		BizChangefeedID: pipeline.Children.ChangefeedID,
	})
	if err != nil {
		return err
	}
	if response == nil || response.BizChangefeed == nil {
		return fmt.Errorf("Changefeed observation is empty")
	}
	changefeed := response.BizChangefeed
	// The child owns the specification once ScaleDataPipeline delegates to it, so
	// repair the Pipeline read model if the scale succeeded but the Pipeline row
	// save that follows it did not.
	if changefeed.SpecificationCode != "" &&
		changefeed.SpecificationCode != pipeline.ChangefeedSpecification {
		pipeline.ChangefeedSpecification = changefeed.SpecificationCode
	}
	observation.ChangefeedStatus = changefeed.GetConnectorStatus().String()
	observation.ChangefeedErrorMessage = sanitizeChildMessage(
		changefeed.ErrorMessage, pipeline.Params.Staging.S3.AccessKeyID)
	if changefeed.CheckpointTSO < observation.ChangefeedCheckpointTSO {
		return fmt.Errorf("Changefeed checkpoint regressed")
	}
	if changefeed.CheckpointTSO != 0 {
		observation.ChangefeedCheckpointTSO = changefeed.CheckpointTSO
		checkpointAt := changefeed.GetCheckpointUTCTime().UTC()
		observation.ChangefeedCheckpointAt = &checkpointAt
	}
	return nil
}

func (m *observationModel) observeLake(
	ctx context.Context,
	pipeline *entity.DataPipeline,
	observation *entity.DataPipelineObservation,
	now time.Time,
) (bool, error) {
	if m.lake == nil {
		return false, fmt.Errorf("Lake observer is unavailable")
	}
	invocation := lakeInvocation(pipeline)
	if _, getErr := m.lake.GetDataSource(ctx, invocation, pipeline.Children.LakeDataSourceName); getErr != nil {
		return false, getErr
	}
	observation.LakeDataSourceStatus = "READY"
	taskObservation, err := m.lake.ObserveTask(ctx, invocation, pipeline.Children.LakeTaskName)
	if err != nil {
		return false, err
	}
	observation.LakeIntegrationStatus = string(taskObservation.TaskStatus)
	observation.LakeErrorMessage = ""
	if lakeRunTerminallyFailed(taskObservation) {
		observation.LakeErrorMessage = destinationRunFailureMessage(taskObservation, pipeline)
		return false, errTerminalDestinationRun
	}
	if taskObservation.LastResolvedTS < observation.LakeResolvedTSO {
		return false, fmt.Errorf("Lake resolved TSO regressed")
	}
	if taskObservation.LastResolvedTS != 0 {
		observation.LakeResolvedTSO = taskObservation.LastResolvedTS
		resolvedAt := taskObservation.LastResolvedAt.UTC()
		if resolvedAt.IsZero() && taskObservation.LastResolvedTS != 0 {
			resolvedAt = tsoTime(taskObservation.LastResolvedTS)
		}
		if resolvedAt.IsZero() {
			resolvedAt = now
		}
		observation.LakeResolvedAt = &resolvedAt
	}
	// A completed full sync stays completed. The destination reports it on the
	// snapshot run, which is the oldest run of a continuous task and eventually
	// falls outside the window a single observation reads; overwriting the
	// stored answer would then walk the Pipeline back to "still loading" long
	// after it finished. The resolved TSO above is guarded the same way.
	observation.LakeFullSyncCompleted = observation.LakeFullSyncCompleted ||
		taskObservation.Progress.FullSyncCompleted
	// Unlike the two facts above, these are taken as reported rather than
	// carried forward. They belong to the run they came from, so holding on to
	// an older run's numbers would describe work that is no longer happening.
	observation.RowsSynced = taskObservation.Progress.RowsSynced
	observation.InsertCount = taskObservation.Progress.InsertCount
	observation.UpdateCount = taskObservation.Progress.UpdateCount
	observation.DeleteCount = taskObservation.Progress.DeleteCount
	observation.TablesActive = taskObservation.Progress.TablesActive
	observation.TablesError = taskObservation.Progress.TablesError
	observation.DatabasesActive = taskObservation.Progress.DatabasesActive
	return observation.LakeFullSyncCompleted, nil
}

// isLiveStatus reports the states a destination failure can terminate. Paused
// and Deleting are excluded: a stopped task is the expected state there, and
// Failed and Completed are already final.
func isLiveStatus(status entity.DataPipelineStatus) bool {
	return status == entity.DataPipelineStatusCreating ||
		status == entity.DataPipelineStatusRunning ||
		status == entity.DataPipelineStatusResuming
}

// errTerminalDestinationRun marks a destination whose latest run failed and
// which nothing will move again, as opposed to a transport error that a later
// poll may resolve.
var errTerminalDestinationRun = errors.New("Lake integration run failed")

// destinationRunFailureMessage says why the destination stopped, in the words
// Lake used. "Lake integration run failed" tells a user that something is
// wrong and nothing about what to change; the run message is the difference
// between that and "warehouse dp_e2e_wh not found", which names the fix.
//
// The message is sanitized like every other child message: these runs are
// configured with the staging credentials, and a downstream that echoes its
// own request must not turn into a leak here.
func destinationRunFailureMessage(
	observed lake.TaskObservation, pipeline *entity.DataPipeline,
) string {
	message := sanitizeChildMessage(
		observed.LatestRunMessage, pipeline.Params.Staging.S3.AccessKeyID)
	if strings.TrimSpace(message) == "" {
		return "Lake integration run failed"
	}
	return "Lake integration run failed: " + message
}

// lakeRunTerminallyFailed reports a destination that has stopped with a failed
// run. Both halves matter: a failed run under a task that is still running may
// still be retried by Lake, and a stopped task whose last run succeeded is the
// ordinary end of a snapshot-only load.
func lakeRunTerminallyFailed(observed lake.TaskObservation) bool {
	return observed.LatestRunStatus == lake.TaskRunStatusFail &&
		observed.TaskStatus == lake.TaskStatusStopped
}

func calculateLatencySeconds(changefeedCheckpointTSO, lakeResolvedTSO uint64) int64 {
	if changefeedCheckpointTSO == 0 || lakeResolvedTSO == 0 {
		return 0
	}
	checkpointMillis := changefeedCheckpointTSO >> tsoPhysicalShiftBits
	resolvedMillis := lakeResolvedTSO >> tsoPhysicalShiftBits
	if checkpointMillis <= resolvedMillis {
		return 0
	}
	return int64((checkpointMillis - resolvedMillis) / 1000)
}