package scale_operation

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"

	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/pkgv2/db"
	"google.golang.org/protobuf/types/known/timestamppb"
	"gorm.io/gorm"
)

const defaultListLimit = 100

type RepoImpl struct {
	provider db.Provider
}

func NewScaleOperationRepo(provider db.Provider) repo.ScaleOperationRepo {
	return &RepoImpl{provider: provider}
}

func (r *RepoImpl) Create(ctx context.Context, operation *entity.ScaleOperation) error {
	if operation == nil {
		return fmt.Errorf("scale operation is nil")
	}
	row, err := entityToSchema(operation)
	if err != nil {
		return err
	}
	if err := r.provider.WriteDB(ctx).Create(row).Error; err != nil {
		return fmt.Errorf("create scale operation (%s/%s): %w", operation.GetSharedPoolId(), operation.GetScaleOperationId(), err)
	}
	return nil
}

func (r *RepoImpl) LoadByID(ctx context.Context, sharedPoolID, scaleOperationID string, includeDeleted bool) (*entity.ScaleOperation, error) {
	query := r.provider.ReadDB(ctx)
	if includeDeleted {
		query = query.Unscoped()
	}

	var row schema.ScaleOperation
	if err := query.Where(
		"shared_pool_id = ? AND scale_operation_id = ?",
		sharedPoolID,
		scaleOperationID,
	).First(&row).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, gorm.ErrRecordNotFound
		}
		return nil, fmt.Errorf("load scale operation (%s/%s): %w", sharedPoolID, scaleOperationID, err)
	}
	return schemaToEntity(&row)
}

// Save fully replaces the row's mutable fields, including zero values (e.g.
// cancelable flipping true -> false). It loads the existing row's id first
// (single read), then issues a single map-based Updates (single write): a
// map, unlike a struct, is never skipped for zero values, and it avoids
// gorm.Save's own existence check on top of the read we already did.
// UpdatedAt is bumped by gorm's autoUpdateTime hook; CreatedAt/DeletedAt are
// intentionally excluded from the map so they are left untouched.
func (r *RepoImpl) Save(ctx context.Context, operation *entity.ScaleOperation) error {
	if operation == nil {
		return fmt.Errorf("scale operation is nil")
	}
	row, err := entityToSchema(operation)
	if err != nil {
		return err
	}
	var existing schema.ScaleOperation
	if err := r.provider.ReadDB(ctx).Select("id").Where(
		"shared_pool_id = ? AND scale_operation_id = ?",
		row.SharedPoolID,
		row.ScaleOperationID,
	).First(&existing).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return gorm.ErrRecordNotFound
		}
		return fmt.Errorf("load scale operation for save (%s/%s): %w", row.SharedPoolID, row.ScaleOperationID, err)
	}
	updates := map[string]any{
		"component_type":      row.ComponentType,
		"status":              row.Status,
		"cancelable":          row.Cancelable,
		"original_count":      row.OriginalCount,
		"target_count":        row.TargetCount,
		"ready_count":         row.ReadyCount,
		"pending_count":       row.PendingCount,
		"result_summary":      row.ResultSummary,
		"original_quota_json": row.OriginalQuotaJSON,
		"evidence_json":       row.EvidenceJSON,
		"completed_at":        row.CompletedAt,
	}
	if err := r.provider.WriteDB(ctx).
		Model(&schema.ScaleOperation{}).
		Where("id = ?", existing.ID).
		Updates(updates).Error; err != nil {
		return fmt.Errorf("save scale operation (%s/%s): %w", row.SharedPoolID, row.ScaleOperationID, err)
	}
	return nil
}

// FindActiveByComponent is unbounded on purpose: the active-lock invariant
// (at most one non-terminal operation per shared_pool_id/component_id) means
// the result set is expected to be tiny. Capping it here would silently mask
// a violation of that invariant instead of surfacing it.
func (r *RepoImpl) FindActiveByComponent(ctx context.Context, sharedPoolID, componentID string) ([]*entity.ScaleOperation, error) {
	var rows []*schema.ScaleOperation
	if err := r.provider.ReadDB(ctx).
		Where("shared_pool_id = ? AND component_id = ? AND status IN ?", sharedPoolID, componentID, []string{
			statusToDB(apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_RUNNING),
			statusToDB(apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_CANCELING),
		}).
		Order("updated_at desc").
		Find(&rows).Error; err != nil {
		return nil, fmt.Errorf("find active scale operations by component: %w", err)
	}
	return rowsToEntities(rows)
}

func (r *RepoImpl) ListNonTerminal(ctx context.Context, limit int) ([]*entity.ScaleOperation, error) {
	if limit <= 0 || limit > defaultListLimit {
		limit = defaultListLimit
	}
	var rows []*schema.ScaleOperation
	if err := r.provider.ReadDB(ctx).
		Where("status IN ?", []string{
			statusToDB(apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_RUNNING),
			statusToDB(apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_CANCELING),
		}).
		Order("updated_at asc").
		Limit(limit).
		Find(&rows).Error; err != nil {
		return nil, fmt.Errorf("list non-terminal scale operations: %w", err)
	}
	return rowsToEntities(rows)
}

func rowsToEntities(rows []*schema.ScaleOperation) ([]*entity.ScaleOperation, error) {
	operations := make([]*entity.ScaleOperation, 0, len(rows))
	for _, row := range rows {
		operation, err := schemaToEntity(row)
		if err != nil {
			return nil, err
		}
		operations = append(operations, operation)
	}
	return operations, nil
}

func schemaToEntity(row *schema.ScaleOperation) (*entity.ScaleOperation, error) {
	if row == nil {
		return nil, fmt.Errorf("scale operation row is nil")
	}
	evidence, err := unmarshalEvidence(row.EvidenceJSON)
	if err != nil {
		return nil, fmt.Errorf("unmarshal scale operation evidence (%s/%s): %w", row.SharedPoolID, row.ScaleOperationID, err)
	}

	operation := &entity.ScaleOperation{
		ScaleOperation: apiv1.ScaleOperation{
			SharedPoolId:     row.SharedPoolID,
			ComponentId:      row.ComponentID,
			ScaleOperationId: row.ScaleOperationID,
			ComponentType:    apiv1.ComponentType(row.ComponentType),
			Status:           statusFromDB(row.Status),
			Cancelable:       row.Cancelable,
			OriginalCount:    row.OriginalCount,
			TargetCount:      row.TargetCount,
			ReadyCount:       row.ReadyCount,
			PendingCount:     row.PendingCount,
			ResultSummary:    row.ResultSummary,
			CreatedAt:        timestamppb.New(row.CreatedAt),
			UpdatedAt:        timestamppb.New(row.UpdatedAt),
		},
		OriginalQuotaJSON: row.OriginalQuotaJSON,
		Evidence:          evidence,
	}
	if row.CompletedAt != nil {
		operation.CompletedAt = timestamppb.New(*row.CompletedAt)
	}
	return operation, nil
}

func entityToSchema(operation *entity.ScaleOperation) (*schema.ScaleOperation, error) {
	evidenceJSON, err := marshalEvidence(operation.Evidence)
	if err != nil {
		return nil, fmt.Errorf("marshal scale operation evidence: %w", err)
	}

	// status maps to a NOT NULL enum column; fail fast on an unknown/unspecified
	// value instead of writing an empty string and surfacing a low-signal DB error.
	status := statusToDB(operation.GetStatus())
	if status == "" {
		return nil, fmt.Errorf("scale operation %s has invalid status %v", operation.GetScaleOperationId(), operation.GetStatus())
	}

	row := &schema.ScaleOperation{
		SharedPoolID:      operation.GetSharedPoolId(),
		ComponentID:       operation.GetComponentId(),
		ScaleOperationID:  operation.GetScaleOperationId(),
		ComponentType:     int32(operation.GetComponentType()),
		Status:            status,
		Cancelable:        operation.GetCancelable(),
		OriginalCount:     operation.GetOriginalCount(),
		TargetCount:       operation.GetTargetCount(),
		ReadyCount:        operation.GetReadyCount(),
		PendingCount:      operation.GetPendingCount(),
		ResultSummary:     operation.GetResultSummary(),
		OriginalQuotaJSON: operation.OriginalQuotaJSON,
		EvidenceJSON:      evidenceJSON,
	}
	if operation.GetCompletedAt() != nil {
		completedAt := operation.GetCompletedAt().AsTime()
		row.CompletedAt = &completedAt
	}
	return row, nil
}

// Evidence is internal-only diagnostic state (never a public proto field), so
// it round-trips through plain JSON rather than protojson.
func marshalEvidence(evidence []*entity.ScaleOperationEvidence) (string, error) {
	if len(evidence) == 0 {
		return "", nil
	}
	raw, err := json.Marshal(evidence)
	if err != nil {
		return "", err
	}
	return string(raw), nil
}

func unmarshalEvidence(raw string) ([]*entity.ScaleOperationEvidence, error) {
	if raw == "" {
		return nil, nil
	}
	var evidence []*entity.ScaleOperationEvidence
	if err := json.Unmarshal([]byte(raw), &evidence); err != nil {
		return nil, err
	}
	return evidence, nil
}

func statusToDB(v entity.ScaleOperationStatus) string {
	switch v {
	case apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_RUNNING:
		return "running"
	case apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_SUCCEEDED:
		return "succeeded"
	case apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_CANCELING:
		return "canceling"
	case apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_CANCELED:
		return "canceled"
	case apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_CANCEL_FAILED:
		return "cancel_failed"
	default:
		return ""
	}
}

func statusFromDB(v string) entity.ScaleOperationStatus {
	switch v {
	case "running":
		return apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_RUNNING
	case "succeeded":
		return apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_SUCCEEDED
	case "canceling":
		return apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_CANCELING
	case "canceled":
		return apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_CANCELED
	case "cancel_failed":
		return apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_CANCEL_FAILED
	default:
		return apiv1.ScaleOperationStatus_SCALE_OPERATION_STATUS_UNSPECIFIED
	}
}