package db

import (
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"sort"
	"strings"

	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"
	"google.golang.org/protobuf/encoding/protojson"
	"gorm.io/gorm"
)

// 20260713002: seed the DEFAULT scope autoscale policy bundle.
//
// Before this migration the DEFAULT scope row in
// autoscale_policy_bundle_specs was empty, so
// GetEffectiveAutoscalePolicy had no baseline for chain resolution.
// All policy decisions fell through to the server-default
// ru_configs values (frozen at shared-pool creation time), or
// were driven purely by HPA behavior for components that have
// an HPA.
//
// The seeded defaults are conservative-safe:
//
//   - All worker components (TIDB_WORKER, TIKV_WORKER,
//     COPROCESSOR_WORKER): SCALE_IN + SCALE_OUT enabled.
//   - TIDB / TIPROXY: SCALE_OUT enabled.
//   - TIDB: SCALE_IN enabled.
//   - TIPROXY: SCALE_IN disabled (HPA-only — no topology-aware
//     scale-in path for tiproxy; SCALE_OUT is the supported
//     direction).
//   - TIKV: SCALE_OUT (CPU), DISK_SIZE expand, IOPS expand,
//     THROUGHPUT expand enabled.
//
// The seeded row has ApplyToAllSharedPools = NULL (opt-in only
// at the TENANT scope), so this DEFAULT row does NOT auto-fanout
// to every shared pool. It only participates in
// GetEffectiveAutoscalePolicy chain resolution as the lowest-
// priority layer.
//
// Re-runnable: ON DUPLICATE KEY UPDATE bumps updated_at AND
// overwrites specs_json + bundle_hash — any prior row inserted
// by an older version of this migration (which may have written
// the JSON in a now-incompatible format, e.g. the pre-rename
// `componentType` field or the pre-prefix `TIKV` enum value
// before the proto switched to `targetType` / `TARGET_TIKV`)
// is healed on the next migration run. The DEFAULT-scope row
// is the immutable baseline; operator customization lives on
// TENANT-scope rows (see PriorityResolveForSharedPoolFromBundle),
// so overwriting the DEFAULT row on re-run is safe — there is
// no operator surface that mutates this row.
//
// The (scope, subject_id) unique key (autoscale_policy_bundle_specs
// index `bundle_scope_subject`) is preserved so that:
//   - Fresh installs AutoMigrate + INSERT as before.
//   - Re-runs over a stale-seed row UPDATE in place (no orphan
//     rows accumulate).
//   - Normal RPC-time UpsertBundle (which writes other scopes
//     like TENANT/SCOPE_SHARED_POOL/SCOPE_TIDB_GROUP under different
//     subject_id values) is unaffected — its (scope, subject_id)
//     key never collides with the DEFAULT seed's
//     ('default', '__default__') tuple.
//
// Specs are built via protojson.Marshal so the JSON format
// exactly matches what the v2 RPC marshal path produces —
// hand-crafted JSON strings diverge in subtle ways (whitespace,
// field ordering) and silently produce empty specs after
// unmarshal. bundle_hash is computed via the same canonical
// (sorted, protojson-encoded) hash function the runtime
// SYNC KEYS_ONLY change-detection path uses
// (domain/server/autoscale_policy_bundle.go::hashBundle), inlined
// here as hashSeedSpecs so the migration does not pull in a new
// package import dependency. The seeded row compares equal to
// its own seed bytes — the runtime "stale" check won't
// false-positive.
//
// created_at / updated_at / bundle_hash are written explicitly
// because AutoscalePolicyBundleSpec does not use gorm.Model
// (the table intentionally drops DeletedAt soft-delete to avoid
// unique-index collisions with soft-deleted rows), so GORM's
// autoCreateTime / autoUpdateTime magic does not apply.
func init() {
	AddDBMigrateRecord("20260713002", func(db *gorm.DB) error {
		// AutoMigrate ensures the table exists. If migration
		// 20260713001 was skipped or never ran, this step
		// creates the table.
		if err := db.AutoMigrate(&schema.AutoscalePolicyBundleSpec{}); err != nil {
			return err
		}

		// Build specs via protojson so the JSON format exactly
		// matches what the v2 RPC marshal path produces.
		policies := []*entity.AutoscalePolicySpec{
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIDB, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIPROXY, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIPROXY, Direction: apiv1.AutoscalePolicy_SCALE_IN, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_DISABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIDB, Direction: apiv1.AutoscalePolicy_SCALE_IN, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER, Direction: apiv1.AutoscalePolicy_SCALE_IN, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER, Direction: apiv1.AutoscalePolicy_SCALE_IN, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER, Direction: apiv1.AutoscalePolicy_SCALE_IN, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_NODE_COUNT, Trigger: apiv1.AutoscalePolicy_CPU, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_DISK_SIZE, Trigger: apiv1.AutoscalePolicy_DISK, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_IOPS, Trigger: apiv1.AutoscalePolicy_IO, State: apiv1.AutoscalePolicy_ENABLED},
			{TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV, Direction: apiv1.AutoscalePolicy_SCALE_OUT, Capability: apiv1.AutoscalePolicy_THROUGHPUT, Trigger: apiv1.AutoscalePolicy_IO, State: apiv1.AutoscalePolicy_ENABLED},
		}
		parts := make([]string, 0, len(policies))
		for _, spec := range policies {
			raw, err := protojson.Marshal(spec)
			if err != nil {
				return fmt.Errorf("marshal DEFAULT spec: %w", err)
			}
			parts = append(parts, string(raw))
		}
		specsJSON := "[" + strings.Join(parts, ",") + "]"

		bundleHash := hashSeedSpecs(policies)

		// ON DUPLICATE KEY UPDATE rewrites specs_json +
		// bundle_hash (not just updated_at) so that re-running
		// the migration against a row inserted by an older,
		// pre-rename version of this migration heals the data
		// in place. See the file-header comment for the full
		// rationale. The first INSERT supplies all columns
		// because the table does not use gorm.Model; created_at
		// / updated_at / bundle_hash have no GORM magic and
		// would otherwise default to '' / 0001-01-01, which
		// breaks both GORM scans and the runtime SYNC
		// KEYS_ONLY stale-check.
		return db.Exec(`
			INSERT INTO autoscale_policy_bundle_specs
				(scope, subject_id, specs_json, bundle_hash, created_at, updated_at)
			VALUES ('default', '__default__', ?, ?, NOW(), NOW())
			ON DUPLICATE KEY UPDATE
				specs_json = VALUES(specs_json),
				bundle_hash = VALUES(bundle_hash),
				updated_at = NOW()
		`, specsJSON, bundleHash).Error
	})
}

// hashSeedSpecs computes the same canonical bundle_hash that the
// runtime SYNC KEYS_ONLY path computes in
// domain/server/autoscale_policy_bundle.go::hashBundle, so the
// value written by this migration is what the runtime will
// reproduce when it reads the row back. If this drifts from the
// runtime's hashBundle, the runtime treats the freshly-seeded
// row as "stale" and the SYNC RPC refuses to accept operator
// edits against it.
//
// Inlined here (rather than imported) to keep the migration
// record package self-contained — adding an import on
// domain/server would also pull in tilog, otel, and the rest of
// the server's transitive deps into a binary that runs before
// the application's runtime logger is wired up.
//
// Algorithm (must match domain/server exactly):
//
//  1. Sort specs by (componentType, direction, capability,
//     trigger). Go map iteration order is randomized, so a
//     request-arrival-order slice would otherwise hash
//     differently each call.
//  2. protojson.Marshal each sorted spec — produces stable
//     text serialization with enum values as string names
//     (matching what the v2 RPC marshal path emits).
//  3. SHA-256 over the concatenated protojson bytes separated by
//     "\n".
//  4. Hex-encode the digest and truncate to 32 chars (fits the
//     column's varchar(64) NOT NULL).
func hashSeedSpecs(specs []*entity.AutoscalePolicySpec) string {
	sorted := make([]*entity.AutoscalePolicySpec, len(specs))
	copy(sorted, specs)
	sort.Slice(sorted, func(i, j int) bool {
		a, b := sorted[i], sorted[j]
		if a.GetTargetType() != b.GetTargetType() {
			return a.GetTargetType() < b.GetTargetType()
		}
		if a.GetDirection() != b.GetDirection() {
			return a.GetDirection() < b.GetDirection()
		}
		if a.GetCapability() != b.GetCapability() {
			return a.GetCapability() < b.GetCapability()
		}
		return a.GetTrigger() < b.GetTrigger()
	})
	h := sha256.New()
	for _, s := range sorted {
		// protojson → stable text serialization independent of map order.
		b, _ := protojson.Marshal(s)
		h.Write(b)
		h.Write([]byte("\n"))
	}
	return hex.EncodeToString(h.Sum(nil))[:32]
}