// Package specdrift provides a generic field-spec drift verifier that
// compares an embedded spec JSON against live _System_Field rows and
// reports Type / Required / Unique / TrackHistory / Default / HelpText / Options / ReferenceTo /
// DeleteConstraint / ProjectionConfig / Validator / ValidatorConfig / ControllingField /
// ChildRelationshipName / SiblingMaxParentField disagreements.
//
// The package is tenant-agnostic: it operates on whatever spec bytes the
// caller hands in. Callers under controlplane/** invoke it for the
// standard spec; callers under controlplane/tenants/** invoke it for
// their own tenant spec. Platform code references no tenant identifier
// here — that contract is what lets the tenant drift check live inside
// the tenant quarantine instead of the platform binary.
package specdrift

import (
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"reflect"
	"sort"
	"strings"

	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-controlplane/packspec"
)

// fieldSpec mirrors the subset of the packspec field schema relevant to
// drift detection. Default is json.RawMessage — not string — so a spec
// entry with a bare JSON number / boolean (e.g. `"default": 0`) does not
// hard-fail at Unmarshal. packspec.CoerceSpecDefault normalizes the raw
// form to the string stored in _System_Field.DefaultValue before
// comparison.
type fieldSpec struct {
	APIName               string                        `json:"apiName"`
	Type                  string                        `json:"type"`
	Createable            *bool                         `json:"createable,omitempty"`
	Updateable            *bool                         `json:"updateable,omitempty"`
	Required              bool                          `json:"required,omitempty"`
	Unique                bool                          `json:"unique,omitempty"`
	TrackHistory          bool                          `json:"trackHistory,omitempty"`
	Default               json.RawMessage               `json:"default,omitempty"`
	HelpText              string                        `json:"inlineHelpText,omitempty"`
	Options               models.PicklistOptions        `json:"options,omitempty"`
	ReferenceTo           []string                      `json:"referenceTo,omitempty"`
	WritableReferenceTo   []string                      `json:"writableReferenceTo,omitempty"`
	DeleteConstraint      string                        `json:"deleteConstraint,omitempty"`
	Projection            *models.FieldProjectionConfig `json:"projection,omitempty"`
	Validator             string                        `json:"validator,omitempty"`
	ValidatorConfig       string                        `json:"validatorConfig,omitempty"`
	ControllingField      string                        `json:"controllingField,omitempty"`
	ChildRelationshipName string                        `json:"childRelationshipName,omitempty"`
	SiblingMaxParentField string                        `json:"siblingMaxParentField,omitempty"`
	WriteProtectionPolicy string                        `json:"writeProtectionPolicy,omitempty"`
	ReadProtectionPolicy  string                        `json:"readProtectionPolicy,omitempty"`
}

type objectSpec struct {
	APIName string      `json:"apiName"`
	Fields  []fieldSpec `json:"fields"`
}

type liveField struct {
	typ                   string
	createable            bool
	updateable            bool
	required              bool
	unique                bool
	trackHistory          bool
	defaultValue          *string
	helpText              *string
	options               *string
	referenceTo           *string
	writableReferenceTo   *string
	deleteRule            *string
	projection            *string
	validator             *string
	validatorConfig       *string
	controllingField      *string
	childRelationshipName *string
	siblingMaxParentField *string
	writeProtectionPolicy *string
	readProtectionPolicy  *string
	isCustomized          bool
}

// CustomizationPolicy controls how the verifier reacts to a live
// _System_Field row whose IsCustomized flag is set. The choice is
// load-bearing for the L1 manageability contract summarized in
// `docs/design/l0-platform-substrate.md` under "Verify":
//
//   - On `locked` surfaces, "tenant/admin edits are drift" — the row
//     flag must NOT downgrade structural drift to a warning.
//   - On `defaults_only` surfaces, IsCustomized=1 is the legitimate
//     ownership marker that says the admin took the live value over;
//     verify must not treat that as drift.
//
// Encoding the choice in the API is what stops the verifier from inventing its
// own override semantics; callers must choose a policy from the pack component
// manageability contract.
type CustomizationPolicy int

const (
	// CustomizationPolicyDrift treats IsCustomized=1 the same as
	// IsCustomized=0: any structural disagreement with spec is fatal
	// drift. Use this for locked surfaces.
	CustomizationPolicyDrift CustomizationPolicy = iota
	// CustomizationPolicyAllow downgrades structural disagreements to
	// warnings when IsCustomized=1, treating the row as admin-owned.
	// Use this for defaults_only surfaces and for the L0 standard-spec
	// path where preserving operator-visible disagreement is the
	// intended audit signal.
	CustomizationPolicyAllow
)

// Verify walks specJSON and logs one warning (or fatal) per field whose
// live _System_Field row disagrees with the spec on Type, Required, Unique, TrackHistory,
// DefaultValue, HelpText, ReferenceTo, DeleteRule, ProjectionConfig, Validator, ValidatorConfig,
// ControllingField, SiblingMaxParentField, or (for picklists) the Options value set.
//
// skipFields is a set of "lower(object).lower(field)" keys the caller
// wants excluded from the comparison — typically fields known to be
// owned by a later-applied spec (e.g. standard verify skips
// tenant-owned fields).
//
// strictMissing controls whether "spec declares this field but no live
// row exists" is fatal. Set true when the caller's apply pass is
// expected to have created every spec row (tenant pack verifying
// itself). Set false when missing rows are a legitimate
// "apply-did-not-run" state (platform verifying its own spec on a fresh
// install; admin soft-deletes).
//
// This wrapper applies the legacy "preserve customized rows as warning"
// policy — operators see the disagreement without verify failing. New
// pack-component callers should call VerifyForTableTypes directly with
// the policy that matches their manageability tier.
func Verify(rawDB *sql.DB, specJSON []byte, specName string, skipFields map[string]bool, strictMissing bool) error {
	return VerifyForTableTypes(
		rawDB,
		specJSON,
		specName,
		skipFields,
		strictMissing,
		CustomizationPolicyAllow,
		constants.TableTypeStandardObject,
	)
}

// VerifyForTableTypes is the same drift verifier scoped to explicit
// _System_Object.TableType values. Platform callers use Verify's
// standard_object default; tenant packs pass custom_object as well when
// they own tenant-created objects under the quarantine package.
//
// customizationPolicy must reflect the manageability tier of the rows
// being verified — see CustomizationPolicy.
func VerifyForTableTypes(rawDB *sql.DB, specJSON []byte, specName string, skipFields map[string]bool, strictMissing bool, customizationPolicy CustomizationPolicy, tableTypes ...constants.TableType) error {
	var specs []objectSpec
	if err := json.Unmarshal(specJSON, &specs); err != nil {
		return fmt.Errorf("parse %s: %w", specName, err)
	}

	tableTypePredicate, tableTypeArgs := tableTypeWhereClause(tableTypes)

	// Scope to the caller-owned table types. Without this filter a custom
	// object whose ApiName collides with a standard name (case-insensitive)
	// could shadow the real row in the live map and fabricate/mask drift.
	query := fmt.Sprintf(
		"SELECT o.`%s`, f.`%s`, "+
			"f.`%s`, f.`%s`, f.`%s`, f.`%s`, "+
			"f.`%s`, f.`%s`, f.`%s`, f.`%s`, "+
			"f.`%s`, f.`%s`, f.`%s`, f.`%s`, f.`%s`, f.`%s`, "+
			"f.`%s`, f.`%s`, f.`%s`, f.`%s`, "+
			"COALESCE(f.`%s`, 0), COALESCE(f.`%s`, 0), COALESCE(f.`%s`, 0) "+
			"FROM `%s` f JOIN `%s` o ON f.`%s` = o.`%s` "+
			"WHERE f.`%s` = 0 AND o.`%s` = 0 AND %s",
		constants.FieldAPIName, constants.FieldAPIName,
		constants.ColType, constants.FieldSysField_Createable, constants.FieldSysField_Updateable, constants.ColRequired,
		"DefaultValue", constants.FieldSysField_HelpText, "Options", constants.FieldSysField_ProjectionConfig,
		constants.FieldReferenceTo, constants.FieldSysField_WritableReferenceTo, constants.FieldSysField_DeleteRule, constants.FieldSysField_Validator, constants.FieldSysField_ValidatorConfig, constants.FieldSysField_ControllingField,
		constants.FieldSysField_ChildRelationshipName,
		constants.FieldSysField_SiblingMaxParentField,
		constants.FieldSysField_WriteProtectionPolicy,
		constants.FieldSysField_ReadProtectionPolicy,
		constants.FieldSysField_IsUnique,
		constants.FieldSysField_TrackHistory,
		constants.FieldSysField_IsCustomized,
		constants.TableField, constants.TableObject,
		constants.FieldObjectID, constants.ColID,
		constants.ColIsDeleted, constants.ColIsDeleted,
		tableTypePredicate,
	)

	rows, err := rawDB.Query(query, tableTypeArgs...)
	if err != nil {
		return fmt.Errorf("query _System_Field for drift: %w", err)
	}
	defer rows.Close()

	live := make(map[string]liveField)
	for rows.Next() {
		var (
			objAPI, fieldAPI, typ string
			createable            bool
			updateable            bool
			required              bool
			unique                bool
			trackHistory          bool
			isCustomized          bool
			defaultValue          sql.NullString
			helpText              sql.NullString
			options               sql.NullString
			projection            sql.NullString
			referenceTo           sql.NullString
			writableReferenceTo   sql.NullString
			deleteRule            sql.NullString
			validator             sql.NullString
			validatorConfig       sql.NullString
			controllingField      sql.NullString
			childRelationshipName sql.NullString
			siblingMaxParentField sql.NullString
			writeProtectionPolicy sql.NullString
			readProtectionPolicy  sql.NullString
		)
		if err := rows.Scan(
			&objAPI, &fieldAPI, &typ, &createable, &updateable, &required, &defaultValue, &helpText, &options, &projection,
			&referenceTo, &writableReferenceTo, &deleteRule, &validator, &validatorConfig, &controllingField, &childRelationshipName, &siblingMaxParentField, &writeProtectionPolicy, &readProtectionPolicy, &unique, &trackHistory, &isCustomized,
		); err != nil {
			return fmt.Errorf("scan drift row: %w", err)
		}
		key := strings.ToLower(objAPI) + "." + strings.ToLower(fieldAPI)
		lr := liveField{typ: typ, createable: createable, updateable: updateable, required: required, unique: unique, trackHistory: trackHistory, isCustomized: isCustomized}
		if defaultValue.Valid {
			v := defaultValue.String
			lr.defaultValue = &v
		}
		if helpText.Valid {
			v := helpText.String
			lr.helpText = &v
		}
		if options.Valid {
			v := options.String
			lr.options = &v
		}
		if referenceTo.Valid {
			v := referenceTo.String
			lr.referenceTo = &v
		}
		if writableReferenceTo.Valid {
			v := writableReferenceTo.String
			lr.writableReferenceTo = &v
		}
		if deleteRule.Valid {
			v := deleteRule.String
			lr.deleteRule = &v
		}
		if projection.Valid {
			v := projection.String
			lr.projection = &v
		}
		if validator.Valid {
			v := validator.String
			lr.validator = &v
		}
		if validatorConfig.Valid {
			v := validatorConfig.String
			lr.validatorConfig = &v
		}
		if controllingField.Valid {
			v := controllingField.String
			lr.controllingField = &v
		}
		if childRelationshipName.Valid {
			v := childRelationshipName.String
			lr.childRelationshipName = &v
		}
		if siblingMaxParentField.Valid {
			v := siblingMaxParentField.String
			lr.siblingMaxParentField = &v
		}
		if writeProtectionPolicy.Valid {
			v := writeProtectionPolicy.String
			lr.writeProtectionPolicy = &v
		}
		if readProtectionPolicy.Valid {
			v := readProtectionPolicy.String
			lr.readProtectionPolicy = &v
		}
		live[key] = lr
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate drift rows: %w", err)
	}

	warnings := 0
	fatals := 0
	missing := 0
	var fatalFieldDrifts []*FieldDriftError
	for _, spec := range specs {
		for _, field := range spec.Fields {
			key := strings.ToLower(spec.APIName) + "." + strings.ToLower(field.APIName)
			if skipFields[key] {
				continue
			}
			lr, ok := live[key]
			if !ok {
				missing++
				log.Printf("   missing: %s.%s (no live _System_Field row)", spec.APIName, field.APIName)
				continue
			}
			diffs := compareFieldAgainstSpec(field, lr)
			if len(diffs) == 0 {
				continue
			}
			isFatal := isStructuralDrift(field, lr) && !customizationPolicyAllowsLiveOverride(customizationPolicy, lr.isCustomized)
			if isFatal {
				fatals += len(diffs)
				fatalFieldDrifts = append(fatalFieldDrifts, &FieldDriftError{
					SpecName:      specName,
					ObjectAPIName: spec.APIName,
					FieldAPIName:  field.APIName,
					Diffs:         append([]FieldDiff(nil), diffs...),
				})
				log.Printf("❌ [FATAL] spec drift: %s.%s", spec.APIName, field.APIName)
			} else {
				warnings += len(diffs)
				log.Printf("⚠️  spec drift: %s.%s", spec.APIName, field.APIName)
			}
			for _, d := range diffs {
				log.Printf("      %s", d.String())
			}
		}
	}
	if missing > 0 {
		if strictMissing {
			fatals += missing
			log.Printf("❌ [FATAL] %s: %d spec-declared field(s) have no live row — pack apply likely failed or was skipped", specName, missing)
		} else {
			log.Printf("ℹ️  %s: %d spec-declared field(s) have no live row — pack apply creates them; soft-deleted rows require manual reconciliation", specName, missing)
		}
	}
	if fatals == 0 && warnings == 0 {
		log.Printf("✅ %s matches live _System_Field (no drift)", specName)
	} else {
		log.Printf("⚠️  %s: %d fatal + %d warning drift(s) detected", specName, fatals, warnings)
	}
	if fatals > 0 {
		if missing > 0 && fatals == missing {
			return &MissingLiveFieldRowsError{SpecName: specName, Count: missing}
		}
		return &StructuralDriftError{
			SpecName:    specName,
			Count:       fatals,
			FieldDrifts: fatalFieldDrifts,
		}
	}
	return nil
}

type FieldDiffKind string

const (
	FieldDiffKindIsUnique         FieldDiffKind = "IsUnique"
	FieldDiffKindTrackHistory     FieldDiffKind = "TrackHistory"
	FieldDiffKindDeleteConstraint FieldDiffKind = "DeleteConstraint"
)

type FieldDiff struct {
	Kind    FieldDiffKind
	Message string
}

func (d FieldDiff) String() string {
	if d.Message != "" {
		return d.Message
	}
	return string(d.Kind)
}

type FieldDriftError struct {
	SpecName      string
	ObjectAPIName string
	FieldAPIName  string
	Diffs         []FieldDiff
}

func (e *FieldDriftError) Error() string {
	if e == nil {
		return "field structural drift"
	}
	return fmt.Sprintf("%s: field drift %s.%s", e.SpecName, e.ObjectAPIName, e.FieldAPIName)
}

func (e *FieldDriftError) IsUniqueOnly() bool {
	return e != nil && e.hasOnlyKinds(FieldDiffKindIsUnique)
}

func (e *FieldDriftError) hasOnlyKinds(kinds ...FieldDiffKind) bool {
	if e == nil || len(e.Diffs) == 0 {
		return false
	}
	allowed := make(map[FieldDiffKind]bool, len(kinds))
	for _, kind := range kinds {
		allowed[kind] = true
	}
	for _, diff := range e.Diffs {
		if !allowed[diff.Kind] {
			return false
		}
	}
	return true
}

type StructuralDriftError struct {
	SpecName    string
	Count       int
	FieldDrifts []*FieldDriftError
}

func (e *StructuralDriftError) Error() string {
	if e == nil {
		return "structural drift(s) — pack apply likely failed or was skipped"
	}
	return fmt.Sprintf("%s: %d structural drift(s) — pack apply likely failed or was skipped", e.SpecName, e.Count)
}

func (e *StructuralDriftError) Unwrap() []error {
	if e == nil || len(e.FieldDrifts) == 0 {
		return nil
	}
	out := make([]error, 0, len(e.FieldDrifts))
	for _, drift := range e.FieldDrifts {
		out = append(out, drift)
	}
	return out
}

func IsUniqueOnlyFieldDrift(err error) bool {
	return hasOnlyFieldDriftKinds(err, FieldDiffKindIsUnique)
}

func IsUniqueOrDeleteConstraintOnlyFieldDrift(err error) bool {
	return hasOnlyFieldDriftKinds(err, FieldDiffKindIsUnique, FieldDiffKindDeleteConstraint)
}

func hasOnlyFieldDriftKinds(err error, kinds ...FieldDiffKind) bool {
	if err == nil {
		return false
	}
	var structural *StructuralDriftError
	if errors.As(err, &structural) {
		if structural == nil || len(structural.FieldDrifts) == 0 {
			return false
		}
		diffCount := 0
		for _, drift := range structural.FieldDrifts {
			if drift == nil || !drift.hasOnlyKinds(kinds...) {
				return false
			}
			diffCount += len(drift.Diffs)
		}
		return diffCount == structural.Count
	}
	var fieldDrift *FieldDriftError
	if errors.As(err, &fieldDrift) {
		return fieldDrift.hasOnlyKinds(kinds...)
	}
	return false
}

type MissingLiveFieldRowsError struct {
	SpecName string
	Count    int
}

func (e *MissingLiveFieldRowsError) Error() string {
	if e == nil {
		return "spec-declared fields have no live rows"
	}
	return fmt.Sprintf("%s: %d structural drift(s) — pack apply likely failed or was skipped", e.SpecName, e.Count)
}

func tableTypeWhereClause(tableTypes []constants.TableType) (string, []interface{}) {
	if len(tableTypes) == 0 {
		tableTypes = []constants.TableType{constants.TableTypeStandardObject}
	}
	args := make([]interface{}, 0, len(tableTypes))
	for _, tableType := range tableTypes {
		args = append(args, string(tableType))
	}
	if len(tableTypes) == 1 {
		return fmt.Sprintf("o.`%s` = ?", constants.FieldSysObject_TableType), args
	}
	placeholders := make([]string, len(tableTypes))
	for i := range placeholders {
		placeholders[i] = "?"
	}
	return fmt.Sprintf("o.`%s` IN (%s)", constants.FieldSysObject_TableType, strings.Join(placeholders, ",")), args
}

// customizationPolicyAllowsLiveOverride encodes the manageability semantics
// for a single _System_Field row: when does its IsCustomized flag legitimately
// downgrade structural drift from fatal to warning?
func customizationPolicyAllowsLiveOverride(policy CustomizationPolicy, isCustomized bool) bool {
	return policy == CustomizationPolicyAllow && isCustomized
}

func effectiveSpecRequired(spec fieldSpec) bool {
	return spec.Required || constants.NormalizeFieldType(spec.Type) == constants.FieldTypeMasterDetail
}

// isStructuralDrift returns true when a spec-owned metadata column differs
// from live state on a non-customized row. These columns are written by pack
// apply; mismatch means apply failed, was skipped, or live metadata drifted.
// Callers must gate customized rows before escalating.
//
// Scope: every column declared in FieldSpec (platform_core.go) that pack
// apply writes. _System_Field columns NOT declared by FieldSpec today
// (MaxLength, MinLength, Regex, RegexMessage, MinValue, MaxValue, Indexed)
// are not covered because pack apply does not author them.
// Until then, those columns can be edited directly in _System_Field without
// surfacing as drift.
func isStructuralDrift(spec fieldSpec, live liveField) bool {
	if !strings.EqualFold(spec.Type, live.typ) {
		return true
	}
	if spec.Createable != nil && *spec.Createable != live.createable {
		return true
	}
	if spec.Updateable != nil && *spec.Updateable != live.updateable {
		return true
	}
	if effectiveSpecRequired(spec) != live.required {
		return true
	}
	if spec.Unique != live.unique {
		return true
	}
	if spec.TrackHistory != live.trackHistory {
		return true
	}
	specDefault, specHasDefault := packspec.CoerceSpecDefault(spec.Default)
	if !defaultValuesMatch(specDefault, specHasDefault, live.defaultValue) {
		return true
	}
	if !stringMetadataMatches(spec.HelpText, live.helpText) {
		return true
	}
	if strings.EqualFold(spec.Type, string(constants.FieldTypePicklist)) && len(spec.Options) > 0 {
		if live.options == nil || *live.options == "" || *live.options == "[]" || *live.options == "null" {
			return true
		}
		if !picklistOptionsMatch(spec.Options, *live.options) {
			return true
		}
	}
	if !referenceToMatches(spec.ReferenceTo, live.referenceTo) {
		return true
	}
	if !referenceToMatches(spec.WritableReferenceTo, live.writableReferenceTo) {
		return true
	}
	if !deleteConstraintMatches(spec.DeleteConstraint, live.deleteRule) {
		return true
	}
	if !projectionConfigsMatch(spec.Projection, live.projection) {
		return true
	}
	if !stringMetadataMatches(spec.Validator, live.validator) {
		return true
	}
	if !stringMetadataMatches(spec.ValidatorConfig, live.validatorConfig) {
		return true
	}
	if !stringMetadataMatches(spec.ControllingField, live.controllingField) {
		return true
	}
	if !stringMetadataMatches(spec.ChildRelationshipName, live.childRelationshipName) {
		return true
	}
	if !stringMetadataMatches(spec.SiblingMaxParentField, live.siblingMaxParentField) {
		return true
	}
	if !stringMetadataMatches(spec.WriteProtectionPolicy, live.writeProtectionPolicy) {
		return true
	}
	if !stringMetadataMatches(spec.ReadProtectionPolicy, live.readProtectionPolicy) {
		return true
	}
	return false
}

// picklistOptionsMatch compares the full pack-authored PicklistOption
// structure — Value, Label, Tags, and opaque Extensions — between spec and live.
// Comparing only Value was the legacy shape; adding Tags in the
// previous pass closed the tenant-overlay classification gap; adding
// Label here closes the remaining asymmetry. Label flows into
// describe responses (sobject_field_helpers.go emits opt.Label) and renders in
// the frontend. Extensions stay opaque to L0, but Apply writes them unchanged,
// so Verify must compare them as part of the same source-of-truth payload.
func picklistOptionsMatch(specOptions models.PicklistOptions, liveJSON string) bool {
	var liveOptions models.PicklistOptions
	if err := json.Unmarshal([]byte(liveJSON), &liveOptions); err != nil {
		return false
	}
	if len(specOptions) != len(liveOptions) {
		return false
	}
	specByValue := indexPicklistByValue(specOptions)
	liveByValue := indexPicklistByValue(liveOptions)
	for value, specOpt := range specByValue {
		liveOpt, ok := liveByValue[value]
		if !ok {
			return false
		}
		if specOpt.Label != liveOpt.Label {
			return false
		}
		if !stringSliceSetEqual(specOpt.Tags, liveOpt.Tags) {
			return false
		}
		if !rawMessageMapEqual(specOpt.Extensions, liveOpt.Extensions) {
			return false
		}
	}
	return true
}

func indexPicklistByValue(opts models.PicklistOptions) map[string]models.PicklistOption {
	byValue := make(map[string]models.PicklistOption, len(opts))
	for _, o := range opts {
		byValue[o.Value] = o
	}
	return byValue
}

func rawMessageMapEqual(a, b map[string]json.RawMessage) bool {
	if len(a) != len(b) {
		return false
	}
	if len(a) == 0 {
		return true
	}
	for key, aValue := range a {
		bValue, ok := b[key]
		if !ok {
			return false
		}
		if !rawJSONEqual(aValue, bValue) {
			return false
		}
	}
	return true
}

func rawJSONEqual(a, b json.RawMessage) bool {
	var aAny any
	var bAny any
	if err := json.Unmarshal(a, &aAny); err != nil {
		return string(a) == string(b)
	}
	if err := json.Unmarshal(b, &bAny); err != nil {
		return string(a) == string(b)
	}
	return reflect.DeepEqual(aAny, bAny)
}

// stringSliceSetEqual compares two tag slices as unordered sets.
// nil and empty slice are treated as equivalent (the "no tags" state).
func stringSliceSetEqual(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	if len(a) == 0 {
		return true
	}
	aCopy := append([]string(nil), a...)
	bCopy := append([]string(nil), b...)
	sort.Strings(aCopy)
	sort.Strings(bCopy)
	for i := range aCopy {
		if aCopy[i] != bCopy[i] {
			return false
		}
	}
	return true
}

func compareFieldAgainstSpec(spec fieldSpec, live liveField) []FieldDiff {
	var diffs []FieldDiff

	if !strings.EqualFold(spec.Type, live.typ) {
		diffs = append(diffs, fieldDiff("Type", fmt.Sprintf("Type: spec=%q live=%q", spec.Type, live.typ)))
	}
	if spec.Createable != nil && *spec.Createable != live.createable {
		diffs = append(diffs, fieldDiff("Createable", fmt.Sprintf("Createable: spec=%t live=%t", *spec.Createable, live.createable)))
	}
	if spec.Updateable != nil && *spec.Updateable != live.updateable {
		diffs = append(diffs, fieldDiff("Updateable", fmt.Sprintf("Updateable: spec=%t live=%t", *spec.Updateable, live.updateable)))
	}
	specRequired := effectiveSpecRequired(spec)
	if specRequired != live.required {
		diffs = append(diffs, fieldDiff("Required", fmt.Sprintf("Required: spec=%t live=%t", specRequired, live.required)))
	}
	if spec.Unique != live.unique {
		diffs = append(diffs, fieldDiff(FieldDiffKindIsUnique, fmt.Sprintf("IsUnique: spec=%t live=%t", spec.Unique, live.unique)))
	}
	if spec.TrackHistory != live.trackHistory {
		diffs = append(diffs, fieldDiff(FieldDiffKindTrackHistory, fmt.Sprintf("TrackHistory: spec=%t live=%t", spec.TrackHistory, live.trackHistory)))
	}
	specDefault, specHasDefault := packspec.CoerceSpecDefault(spec.Default)
	if !defaultValuesMatch(specDefault, specHasDefault, live.defaultValue) {
		diffs = append(diffs, fieldDiff("DefaultValue", fmt.Sprintf("DefaultValue: spec=%s live=%s",
			specDefaultDisplay(specDefault, specHasDefault),
			nullStringDisplay(live.defaultValue))))
	}
	if diff := compareStringMetadata("HelpText", spec.HelpText, live.helpText); diff != "" {
		diffs = append(diffs, fieldDiff("HelpText", diff))
	}
	if strings.EqualFold(spec.Type, string(constants.FieldTypePicklist)) {
		if diff := comparePicklistOptions(spec.Options, live.options); diff != "" {
			diffs = append(diffs, fieldDiff("Options", diff))
		}
	}
	if diff := compareReferenceTo(spec.ReferenceTo, live.referenceTo); diff != "" {
		diffs = append(diffs, fieldDiff("ReferenceTo", diff))
	}
	if diff := compareReferenceSet("WritableReferenceTo", spec.WritableReferenceTo, live.writableReferenceTo); diff != "" {
		diffs = append(diffs, fieldDiff("WritableReferenceTo", diff))
	}
	if diff := compareDeleteConstraint(spec.DeleteConstraint, live.deleteRule); diff != "" {
		diffs = append(diffs, fieldDiff(FieldDiffKindDeleteConstraint, diff))
	}
	if diff := compareProjectionConfig(spec.Projection, live.projection); diff != "" {
		diffs = append(diffs, fieldDiff("ProjectionConfig", diff))
	}
	if diff := compareStringMetadata("Validator", spec.Validator, live.validator); diff != "" {
		diffs = append(diffs, fieldDiff("Validator", diff))
	}
	if diff := compareStringMetadata("ValidatorConfig", spec.ValidatorConfig, live.validatorConfig); diff != "" {
		diffs = append(diffs, fieldDiff("ValidatorConfig", diff))
	}
	if diff := compareStringMetadata("ChildRelationshipName", spec.ChildRelationshipName, live.childRelationshipName); diff != "" {
		diffs = append(diffs, fieldDiff("ChildRelationshipName", diff))
	}
	if diff := compareStringMetadata("SiblingMaxParentField", spec.SiblingMaxParentField, live.siblingMaxParentField); diff != "" {
		diffs = append(diffs, fieldDiff("SiblingMaxParentField", diff))
	}
	if diff := compareStringMetadata("ControllingField", spec.ControllingField, live.controllingField); diff != "" {
		diffs = append(diffs, fieldDiff("ControllingField", diff))
	}
	if diff := compareStringMetadata("WriteProtectionPolicy", spec.WriteProtectionPolicy, live.writeProtectionPolicy); diff != "" {
		diffs = append(diffs, fieldDiff("WriteProtectionPolicy", diff))
	}
	if diff := compareStringMetadata("ReadProtectionPolicy", spec.ReadProtectionPolicy, live.readProtectionPolicy); diff != "" {
		diffs = append(diffs, fieldDiff("ReadProtectionPolicy", diff))
	}
	return diffs
}

func fieldDiff(kind FieldDiffKind, message string) FieldDiff {
	return FieldDiff{Kind: kind, Message: message}
}

func compareReferenceTo(specReferenceTo []string, liveJSON *string) string {
	return compareReferenceSet("ReferenceTo", specReferenceTo, liveJSON)
}

func compareReferenceSet(label string, specReferenceTo []string, liveJSON *string) string {
	if referenceToMatches(specReferenceTo, liveJSON) {
		return ""
	}
	return fmt.Sprintf("%s: spec=%s live=%s", label,
		formatStringSet(specReferenceTo),
		nullStringDisplay(liveJSON))
}

func referenceToMatches(specReferenceTo []string, liveJSON *string) bool {
	if len(specReferenceTo) == 0 {
		return liveJSON == nil || *liveJSON == "" || *liveJSON == "[]" || *liveJSON == "null"
	}
	if liveJSON == nil || *liveJSON == "" || *liveJSON == "null" {
		return false
	}
	var liveReferenceTo []string
	if err := json.Unmarshal([]byte(*liveJSON), &liveReferenceTo); err != nil {
		return false
	}
	return stringSliceSetEqual(specReferenceTo, liveReferenceTo)
}

func compareDeleteConstraint(specValue string, liveValue *string) string {
	if deleteConstraintMatches(specValue, liveValue) {
		return ""
	}
	return fmt.Sprintf("DeleteConstraint: spec=%s live=%s", deleteConstraintDisplay(specValue), deleteConstraintLiveDisplay(liveValue))
}

func deleteConstraintMatches(specValue string, liveValue *string) bool {
	specValue = strings.TrimSpace(specValue)
	if specValue == "" {
		return true
	}
	if liveValue == nil {
		return false
	}
	specRule := constants.NormalizeDeleteRule(constants.DeleteRule(specValue))
	liveRule := constants.NormalizeDeleteRule(constants.DeleteRule(*liveValue))
	return specRule != "" && specRule == liveRule
}

func deleteConstraintDisplay(value string) string {
	if rule := constants.NormalizeDeleteRule(constants.DeleteRule(value)); rule != "" {
		return string(rule)
	}
	if value = strings.TrimSpace(value); value != "" {
		return value
	}
	return "NULL"
}

func deleteConstraintLiveDisplay(value *string) string {
	if value == nil {
		return "NULL"
	}
	return deleteConstraintDisplay(*value)
}

func compareStringMetadata(label string, specValue string, liveValue *string) string {
	if stringMetadataMatches(specValue, liveValue) {
		return ""
	}
	return fmt.Sprintf("%s: spec=%s live=%s", label, stringValueDisplay(specValue), nullStringDisplay(liveValue))
}

func stringMetadataMatches(specValue string, liveValue *string) bool {
	if specValue == "" {
		return liveValue == nil || *liveValue == ""
	}
	if liveValue == nil {
		return false
	}
	return specValue == *liveValue
}

func compareProjectionConfig(specProjection *models.FieldProjectionConfig, liveJSON *string) string {
	if projectionConfigsMatch(specProjection, liveJSON) {
		return ""
	}
	return fmt.Sprintf("ProjectionConfig: spec=%s live=%s",
		projectionDisplay(specProjection),
		nullStringDisplay(liveJSON))
}

func projectionConfigsMatch(specProjection *models.FieldProjectionConfig, liveJSON *string) bool {
	if specProjection == nil {
		return liveJSON == nil || *liveJSON == "" || *liveJSON == "null"
	}
	if liveJSON == nil || *liveJSON == "" || *liveJSON == "null" {
		return false
	}
	var liveProjection models.FieldProjectionConfig
	if err := json.Unmarshal([]byte(*liveJSON), &liveProjection); err != nil {
		return false
	}
	return projectionDisplay(specProjection) == projectionDisplay(&liveProjection)
}

func projectionDisplay(projection *models.FieldProjectionConfig) string {
	if projection == nil {
		return "NULL"
	}
	b, err := json.Marshal(projection)
	if err != nil {
		return "<invalid>"
	}
	return string(b)
}

func defaultValuesMatch(specDefault string, specHasDefault bool, liveDefault *string) bool {
	if !specHasDefault {
		return liveDefault == nil || *liveDefault == ""
	}
	if liveDefault == nil {
		return false
	}
	return specDefault == *liveDefault
}

func specDefaultDisplay(value string, hasDefault bool) string {
	if !hasDefault {
		return "<unset>"
	}
	return fmt.Sprintf("%q", value)
}

func nullStringDisplay(s *string) string {
	if s == nil {
		return "NULL"
	}
	return fmt.Sprintf("%q", *s)
}

func stringValueDisplay(s string) string {
	if s == "" {
		return "<unset>"
	}
	return fmt.Sprintf("%q", s)
}

func comparePicklistOptions(specOpts models.PicklistOptions, liveJSON *string) string {
	var liveOpts models.PicklistOptions
	if liveJSON != nil && *liveJSON != "" {
		_ = json.Unmarshal([]byte(*liveJSON), &liveOpts)
	}
	specValues := sortedValues(specOpts)
	liveValues := sortedValues(liveOpts)
	if strings.Join(specValues, ",") != strings.Join(liveValues, ",") {
		return fmt.Sprintf("Options: spec=[%s] live=[%s]",
			strings.Join(specValues, ","), strings.Join(liveValues, ","))
	}
	// Values match; now check per-value Label, Tags, and Extensions so every
	// field Apply writes to PicklistOption is symmetric with Verify reads.
	if diff := comparePicklistPerValueFields(specOpts, liveOpts); diff != "" {
		return diff
	}
	return ""
}

// comparePicklistPerValueFields returns the first per-value field mismatch
// (Label, Tags, or Extensions) it finds, or "" if every value's fields match.
// Called only after value-set equality is established, so map lookups into
// liveByValue are safe.
func comparePicklistPerValueFields(specOpts, liveOpts models.PicklistOptions) string {
	liveByValue := indexPicklistByValue(liveOpts)
	for _, specOpt := range specOpts {
		liveOpt := liveByValue[specOpt.Value]
		if specOpt.Label != liveOpt.Label {
			return fmt.Sprintf("Label[%s]: spec=%q live=%q",
				specOpt.Value, specOpt.Label, liveOpt.Label)
		}
		if !stringSliceSetEqual(specOpt.Tags, liveOpt.Tags) {
			return fmt.Sprintf("Tags[%s]: spec=%s live=%s",
				specOpt.Value, formatTagSet(specOpt.Tags), formatTagSet(liveOpt.Tags))
		}
		if !rawMessageMapEqual(specOpt.Extensions, liveOpt.Extensions) {
			return fmt.Sprintf("Extensions[%s]: spec=%s live=%s",
				specOpt.Value, formatRawMessageMap(specOpt.Extensions), formatRawMessageMap(liveOpt.Extensions))
		}
	}
	return ""
}

func formatRawMessageMap(values map[string]json.RawMessage) string {
	if len(values) == 0 {
		return "{}"
	}
	raw, err := json.Marshal(values)
	if err != nil {
		return fmt.Sprintf("%v", values)
	}
	return string(raw)
}

func formatTagSet(tags []string) string {
	return formatStringSet(tags)
}

func formatStringSet(values []string) string {
	if len(values) == 0 {
		return "[]"
	}
	sorted := append([]string(nil), values...)
	sort.Strings(sorted)
	return "[" + strings.Join(sorted, ",") + "]"
}

func sortedValues(opts models.PicklistOptions) []string {
	vals := make([]string, 0, len(opts))
	for _, o := range opts {
		vals = append(vals, o.Value)
	}
	sort.Strings(vals)
	return vals
}