package packs

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net/url"
	"sort"
	"strings"
	"time"

	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
)

// connectorCalloutProfileRetryPolicy mirrors the connector_callout.RetryPolicy
// struct for JSON storage purposes. Kept local to avoid a cross-package import
// from the packs layer into the features layer.
type connectorCalloutProfileRetryPolicy struct {
	MaxAttempts          int           `json:"maxAttempts"`
	BaseDelay            time.Duration `json:"baseDelay"`
	Multiplier           float64       `json:"multiplier"`
	MaxDelay             time.Duration `json:"maxDelay"`
	RetryableStatusCodes []int         `json:"retryableStatusCodes,omitempty"`
}

// connectorCalloutProfilePageSpec mirrors connector_callout.PageSpec.
type connectorCalloutProfilePageSpec struct {
	NextCursorJSONPath   string `json:"nextCursorJSONPath"`
	CursorRequestParam   string `json:"cursorRequestParam"`
	PageSizeRequestParam string `json:"pageSizeRequestParam,omitempty"`
	PageSizeValue        int    `json:"pageSizeValue"`
}

// connectorCalloutProfileBatchSpec mirrors connector_callout.BatchSpec.
type connectorCalloutProfileBatchSpec struct {
	Method          string `json:"method"`
	RelativePath    string `json:"relativePath"`
	MaxBatchSize    int    `json:"maxBatchSize"`
	IDsRequestField string `json:"idsRequestField"`
	// RecordsResponsePath is the dotted JSON path to the records array in the
	// batch response body. Empty means the top-level response is the records array.
	RecordsResponsePath string `json:"recordsResponsePath,omitempty"`
	// IDResponseField is the dotted JSON path inside each returned record whose
	// value is the source-ID key used to demux the response.
	IDResponseField string `json:"idResponseField,omitempty"`
	// CredentialRef is the DeveloperName of the NamedCredential used for batch
	// calls. Required when batchSpec is configured: Engine.BatchRead passes this
	// to Throttler.Throttle and CredentialResolver, so an empty value fails at
	// runtime with ErrCredentialUnavailable / ErrEndpointUnavailable.
	CredentialRef string `json:"credentialRef,omitempty"`
	// EndpointClass is the logical endpoint group forwarded to the throttler.
	// Required when batchSpec is configured: Engine.BatchRead passes this to
	// Throttler.Throttle; an empty value causes connector_throttle to fail
	// closed with ErrInvalidEndpointClass.
	EndpointClass string `json:"endpointClass,omitempty"`
	// QueryParams are query-string parameters appended to every ordinary batch
	// request. Values remain arrays so repeated query parameters survive pack
	// parsing, persistence, and drift verification without being collapsed.
	QueryParams url.Values `json:"queryParams,omitempty"`
	// InactiveProbeQueryParams are overlaid on QueryParams only for the bounded
	// one-record confirmation probe performed after an ordinary batch read omits
	// the requested record. The connector pack owns vendor-specific keys; the
	// control plane only preserves and validates the generic query-string shape.
	InactiveProbeQueryParams url.Values `json:"inactiveProbeQueryParams,omitempty"`
	// ExtraBodyFields are additional top-level JSON fields merged into each chunk
	// request body alongside IDsRequestField (e.g. a "properties" projection list).
	// Values must be valid pre-marshaled JSON. Keys colliding with IDsRequestField
	// are silently ignored. Optional.
	ExtraBodyFields map[string]json.RawMessage `json:"extraBodyFields,omitempty"`
	// ProjectionBodyField names an ExtraBodyFields key whose JSON array receives
	// source fields required by installed inbound connector_field_mapping rows.
	// The connector profile owns the vendor request-body key; app/tenant packs own
	// the field inventory through their mappings. Optional.
	ProjectionBodyField string `json:"projectionBodyField,omitempty"`
	// RecordFlattenKey, when non-empty, names a nested object key in each batch
	// response record whose entries are promoted to the top level of the record.
	// Existing top-level keys win; the nested key is removed after promotion.
	// Optional; pack-declared so L0 stays vendor-blind.
	RecordFlattenKey string `json:"recordFlattenKey,omitempty"`
	// AllowCanonicalIDAliases permits a batch read response keyed by a canonical
	// record ID to satisfy the requested source ID when the vendor aliases merged
	// records. Optional; pack-declared because alias semantics are vendor-specific.
	AllowCanonicalIDAliases bool `json:"allowCanonicalIdAliases,omitempty"`
}

// ConnectorCalloutProfileOutboundCapability is one exact provider-authored
// permission for a durable connector push effect. The immutable connector
// binding definition, rather than the Integration runtime, owns this allowlist.
//
// Operation is one of create, update, search, or associate. TargetObject names the
// remote object family. AssociationTargetObject is required only for associate
// capabilities. Method and PathTemplate must match the actual durable contract
// exactly before Catalog publishes the operation.
type ConnectorCalloutProfileOutboundCapability struct {
	Operation               string `json:"operation"`
	TargetObject            string `json:"targetObject"`
	AssociationTargetObject string `json:"associationTargetObject,omitempty"`
	Method                  string `json:"method"`
	PathTemplate            string `json:"pathTemplate"`
}

// connectorCalloutProfileDefinition is the payload shape for a single callout
// profile declared by a connector pack. The definition is pack-owned (L1).
type connectorCalloutProfileDefinition struct {
	// ProfileName is the unique profile name within the owning pack.
	ProfileName string `json:"profileName"`
	// CredentialRef is the DeveloperName of the NamedCredential.
	CredentialRef string `json:"credentialRef"`
	// EndpointClass is the throttle classifier for Engine.Do and Engine.Paginate.
	EndpointClass string `json:"endpointClass"`
	// RetryPolicy configures retry behaviour.
	RetryPolicy connectorCalloutProfileRetryPolicy `json:"retryPolicy"`
	// PageSpec configures cursor-based pagination (optional).
	PageSpec connectorCalloutProfilePageSpec `json:"pageSpec,omitempty"`
	// BatchSpec configures chunked batch reads (optional).
	BatchSpec connectorCalloutProfileBatchSpec `json:"batchSpec,omitempty"`
	// ObjectPathMap maps logical object names to remote API paths (optional).
	ObjectPathMap map[string]string `json:"objectPathMap,omitempty"`
	// OutboundCapabilities is the exact allowlist consumed by Catalog when it
	// joins a durable connector-push contract to this profile. It is deliberately
	// absent from the mutable callout-profile row: the immutable, source-free
	// connector binding definition is its authority.
	OutboundCapabilities []ConnectorCalloutProfileOutboundCapability `json:"outboundCapabilities,omitempty"`
	// Description is an optional admin-readable note.
	Description *string `json:"description,omitempty"`
}

// connectorCalloutProfileProjectionRow is the expected/live projection shape.
// All behavior-defining fields are included so that drift detection catches
// mutations to RetryPolicy, PageSpec, BatchSpec, and ObjectPathMap in addition
// to the identity fields. Any column written by Apply must appear here.
type connectorCalloutProfileProjectionRow struct {
	InstalledPackID string                             `json:"installedPackId,omitempty"`
	OwnerPack       string                             `json:"ownerPack"`
	ProfileName     string                             `json:"profileName"`
	CredentialRef   string                             `json:"credentialRef"`
	EndpointClass   string                             `json:"endpointClass"`
	RetryPolicy     connectorCalloutProfileRetryPolicy `json:"retryPolicy"`
	PageSpec        connectorCalloutProfilePageSpec    `json:"pageSpec"`
	BatchSpec       connectorCalloutProfileBatchSpec   `json:"batchSpec"`
	ObjectPathMap   map[string]string                  `json:"objectPathMap"`
	Description     string                             `json:"description"`
}

// connectorCalloutProfileComponent implements PackComponent for
// ComponentKindConnectorCalloutProfile. It owns exactly one row in
// _System_ConnectorCalloutProfile. The component is locked (admins cannot
// override profile configuration — it is an L1 product fact).
type connectorCalloutProfileComponent struct {
	packName   string
	definition connectorCalloutProfileDefinition
	payload    []byte
}

var _ componentCustomVerifier = (*connectorCalloutProfileComponent)(nil)

type connectorCalloutProfileComponentOptions struct {
	allowLegacyMissingBatchIDResponseField bool
}

type connectorCalloutProfileComponentOption func(*connectorCalloutProfileComponentOptions)

func withLegacyMissingBatchIDResponseFieldCompat() connectorCalloutProfileComponentOption {
	return func(opts *connectorCalloutProfileComponentOptions) {
		opts.allowLegacyMissingBatchIDResponseField = true
	}
}

// NewConnectorCalloutProfileComponent constructs a connectorCalloutProfileComponent
// from a parsed definition.
func NewConnectorCalloutProfileComponent(packName string, def connectorCalloutProfileDefinition) (PackComponent, error) {
	return newConnectorCalloutProfileComponent(packName, def)
}

func newConnectorCalloutProfileComponent(packName string, def connectorCalloutProfileDefinition, componentOpts ...connectorCalloutProfileComponentOption) (PackComponent, error) {
	var opts connectorCalloutProfileComponentOptions
	for _, opt := range componentOpts {
		if opt != nil {
			opt(&opts)
		}
	}
	packName = strings.TrimSpace(packName)
	if packName == "" {
		return nil, fmt.Errorf("connector callout profile component: packName is required")
	}
	def.ProfileName = strings.ToLower(strings.TrimSpace(def.ProfileName))
	if def.ProfileName == "" {
		return nil, fmt.Errorf("connector callout profile component: profileName is required")
	}
	def.CredentialRef = strings.TrimSpace(def.CredentialRef)
	if def.CredentialRef == "" {
		return nil, fmt.Errorf("connector callout profile component: credentialRef is required")
	}
	def.EndpointClass = strings.TrimSpace(def.EndpointClass)
	if def.EndpointClass == "" {
		return nil, fmt.Errorf("connector callout profile component: endpointClass is required")
	}
	if def.RetryPolicy.MaxAttempts < 1 {
		return nil, fmt.Errorf("connector callout profile component: retryPolicy.maxAttempts must be >= 1, got %d", def.RetryPolicy.MaxAttempts)
	}
	if err := validateConnectorCalloutProfileOutboundCapabilities(def.OutboundCapabilities); err != nil {
		return nil, fmt.Errorf("connector callout profile component: outboundCapabilities: %w", err)
	}
	if def.PageSpec.PageSizeValue < 0 {
		return nil, fmt.Errorf("connector callout profile component: pageSpec.pageSizeValue must be >= 0, got %d", def.PageSpec.PageSizeValue)
	}
	// Normalize whitespace on PageSpec inner strings before validation.
	def.PageSpec.NextCursorJSONPath = strings.TrimSpace(def.PageSpec.NextCursorJSONPath)
	def.PageSpec.CursorRequestParam = strings.TrimSpace(def.PageSpec.CursorRequestParam)
	def.PageSpec.PageSizeRequestParam = strings.TrimSpace(def.PageSpec.PageSizeRequestParam)

	// Normalize whitespace on BatchSpec inner strings before validation.
	def.BatchSpec.RelativePath = strings.TrimSpace(def.BatchSpec.RelativePath)
	def.BatchSpec.Method = strings.TrimSpace(def.BatchSpec.Method)
	def.BatchSpec.CredentialRef = strings.TrimSpace(def.BatchSpec.CredentialRef)
	def.BatchSpec.EndpointClass = strings.TrimSpace(def.BatchSpec.EndpointClass)
	def.BatchSpec.IDsRequestField = strings.TrimSpace(def.BatchSpec.IDsRequestField)
	def.BatchSpec.RecordsResponsePath = strings.TrimSpace(def.BatchSpec.RecordsResponsePath)
	def.BatchSpec.IDResponseField = strings.TrimSpace(def.BatchSpec.IDResponseField)
	def.BatchSpec.ProjectionBodyField = strings.TrimSpace(def.BatchSpec.ProjectionBodyField)
	if err := validateConnectorCalloutProfileQueryParams("queryParams", def.BatchSpec.QueryParams); err != nil {
		return nil, fmt.Errorf("connector callout profile component: batchSpec.queryParams invalid: %w", err)
	}
	if err := validateConnectorCalloutProfileQueryParams("inactiveProbeQueryParams", def.BatchSpec.InactiveProbeQueryParams); err != nil {
		return nil, fmt.Errorf("connector callout profile component: batchSpec.inactiveProbeQueryParams invalid: %w", err)
	}

	// PageSpec — any field set means pagination is configured; require all cursor fields.
	pageSpecSet := def.PageSpec.NextCursorJSONPath != "" ||
		def.PageSpec.CursorRequestParam != "" ||
		def.PageSpec.PageSizeRequestParam != "" ||
		def.PageSpec.PageSizeValue != 0
	if pageSpecSet {
		if def.PageSpec.NextCursorJSONPath == "" {
			return nil, fmt.Errorf("connector callout profile component: pageSpec.nextCursorJSONPath is required when pagination is configured")
		}
		if def.PageSpec.CursorRequestParam == "" {
			return nil, fmt.Errorf("connector callout profile component: pageSpec.cursorRequestParam is required when pagination is configured")
		}
	}
	// BatchSpec — any of the 6 required fields OR any optional-but-batch-only field set
	// means batch is configured; require all 6 required fields.
	// ExtraBodyFields, RecordFlattenKey, ProjectionBodyField, and AllowCanonicalIDAliases
	// are optional enrichment, but they only make sense inside a batchSpec; setting any
	// one alone without the required fields would produce a non-executable batchSpec
	// that silently drops enrichment at runtime.
	batchSpecSet := def.BatchSpec.RelativePath != "" ||
		def.BatchSpec.Method != "" ||
		def.BatchSpec.IDsRequestField != "" ||
		def.BatchSpec.MaxBatchSize != 0 ||
		def.BatchSpec.CredentialRef != "" ||
		def.BatchSpec.EndpointClass != "" ||
		def.BatchSpec.RecordsResponsePath != "" ||
		def.BatchSpec.IDResponseField != "" ||
		len(def.BatchSpec.QueryParams) > 0 ||
		len(def.BatchSpec.InactiveProbeQueryParams) > 0 ||
		len(def.BatchSpec.ExtraBodyFields) > 0 ||
		def.BatchSpec.ProjectionBodyField != "" ||
		def.BatchSpec.RecordFlattenKey != "" ||
		def.BatchSpec.AllowCanonicalIDAliases
	if batchSpecSet {
		if def.BatchSpec.RelativePath == "" {
			return nil, fmt.Errorf("connector callout profile component: batchSpec.relativePath is required when batchSpec is configured")
		}
		if def.BatchSpec.Method == "" {
			return nil, fmt.Errorf("connector callout profile component: batchSpec.method is required when batchSpec is configured")
		}
		if def.BatchSpec.IDsRequestField == "" {
			return nil, fmt.Errorf("connector callout profile component: batchSpec.idsRequestField is required when batchSpec is configured")
		}
		if def.BatchSpec.MaxBatchSize < 1 {
			return nil, fmt.Errorf("connector callout profile component: batchSpec.maxBatchSize must be >= 1 when batchSpec is configured, got %d", def.BatchSpec.MaxBatchSize)
		}
		if def.BatchSpec.CredentialRef == "" {
			return nil, fmt.Errorf("connector callout profile component: batchSpec.credentialRef is required when batchSpec is configured")
		}
		if def.BatchSpec.EndpointClass == "" {
			return nil, fmt.Errorf("connector callout profile component: batchSpec.endpointClass is required when batchSpec is configured")
		}
		if def.BatchSpec.IDResponseField == "" {
			if !opts.allowLegacyMissingBatchIDResponseField {
				return nil, fmt.Errorf("connector callout profile component: batchSpec.idResponseField is required when batchSpec is configured")
			}
		} else if err := validateConnectorCalloutProfileDottedPath(def.BatchSpec.IDResponseField); err != nil {
			return nil, fmt.Errorf("connector callout profile component: batchSpec.idResponseField invalid: %w", err)
		}
		if def.BatchSpec.RecordsResponsePath != "" {
			if err := validateConnectorCalloutProfileDottedPath(def.BatchSpec.RecordsResponsePath); err != nil {
				return nil, fmt.Errorf("connector callout profile component: batchSpec.recordsResponsePath invalid: %w", err)
			}
		}
		if def.BatchSpec.ProjectionBodyField != "" {
			if def.BatchSpec.ProjectionBodyField == def.BatchSpec.IDsRequestField {
				return nil, fmt.Errorf("connector callout profile component: batchSpec.projectionBodyField %q must not equal batchSpec.idsRequestField", def.BatchSpec.ProjectionBodyField)
			}
			if err := validateConnectorCalloutProjectionBodyField(def.BatchSpec); err != nil {
				return nil, fmt.Errorf("connector callout profile component: batchSpec.projectionBodyField invalid: %w", err)
			}
			if countExecutableConnectorCalloutObjectPaths(def.ObjectPathMap) != 1 {
				return nil, fmt.Errorf("connector callout profile component: batchSpec.projectionBodyField requires exactly one executable objectPathMap entry in the active-path slice")
			}
		}
	}
	payload, err := json.Marshal(def)
	if err != nil {
		return nil, fmt.Errorf("connector callout profile component: marshal payload: %w", err)
	}
	return &connectorCalloutProfileComponent{
		packName:   packName,
		definition: def,
		payload:    payload,
	}, nil
}

func validateConnectorCalloutProfileQueryParams(fieldName string, values url.Values) error {
	for key, entries := range values {
		if strings.TrimSpace(key) == "" {
			return fmt.Errorf("%s contains a blank key", fieldName)
		}
		if len(entries) == 0 {
			return fmt.Errorf("%s[%q] must contain at least one string value", fieldName, key)
		}
	}
	return nil
}

func validateConnectorCalloutProfileOutboundCapabilities(
	values []ConnectorCalloutProfileOutboundCapability,
) error {
	seen := make(map[string]struct{}, len(values))
	for index := range values {
		value := values[index]
		for _, field := range []struct {
			name string
			raw  string
		}{
			{name: "operation", raw: value.Operation},
			{name: "targetObject", raw: value.TargetObject},
			{name: "method", raw: value.Method},
			{name: "pathTemplate", raw: value.PathTemplate},
		} {
			if strings.TrimSpace(field.raw) == "" || strings.TrimSpace(field.raw) != field.raw {
				return fmt.Errorf("entry %d %s is blank or non-canonical", index, field.name)
			}
		}
		if value.Operation != strings.ToLower(value.Operation) {
			return fmt.Errorf("entry %d operation %q must be lowercase", index, value.Operation)
		}
		if value.Method != strings.ToUpper(value.Method) {
			return fmt.Errorf("entry %d method %q must be uppercase", index, value.Method)
		}
		if !strings.HasPrefix(value.PathTemplate, "/") {
			return fmt.Errorf("entry %d pathTemplate %q must start with '/'", index, value.PathTemplate)
		}
		switch value.Operation {
		case "create":
			if value.Method != "POST" {
				return fmt.Errorf("entry %d create method must be POST", index)
			}
			if value.AssociationTargetObject != "" {
				return fmt.Errorf("entry %d create associationTargetObject must be blank", index)
			}
		case "update":
			if value.Method != "PATCH" {
				return fmt.Errorf("entry %d update method must be PATCH", index)
			}
			if value.AssociationTargetObject != "" {
				return fmt.Errorf("entry %d update associationTargetObject must be blank", index)
			}
		case "search":
			if value.Method != "POST" {
				return fmt.Errorf("entry %d search method must be POST", index)
			}
			if value.AssociationTargetObject != "" {
				return fmt.Errorf("entry %d search associationTargetObject must be blank", index)
			}
		case "associate":
			if strings.TrimSpace(value.AssociationTargetObject) == "" ||
				strings.TrimSpace(value.AssociationTargetObject) != value.AssociationTargetObject {
				return fmt.Errorf("entry %d associate associationTargetObject is blank or non-canonical", index)
			}
		default:
			return fmt.Errorf("entry %d operation %q is unsupported", index, value.Operation)
		}
		identity := strings.Join([]string{
			value.Operation,
			value.TargetObject,
			value.AssociationTargetObject,
			value.Method,
			value.PathTemplate,
		}, "\x00")
		if _, duplicate := seen[identity]; duplicate {
			return fmt.Errorf(
				"entry %d repeats %s capability for %q/%q with %s %q",
				index,
				value.Operation,
				value.TargetObject,
				value.AssociationTargetObject,
				value.Method,
				value.PathTemplate,
			)
		}
		seen[identity] = struct{}{}
	}
	return nil
}

func validateConnectorCalloutProfileDottedPath(path string) error {
	if strings.HasPrefix(path, ".") {
		return fmt.Errorf("must not start with '.'")
	}
	if strings.HasSuffix(path, ".") {
		return fmt.Errorf("must not end with '.'")
	}
	for i, seg := range strings.Split(path, ".") {
		trimmed := strings.TrimSpace(seg)
		if trimmed == "" {
			return fmt.Errorf("segment %d is empty or whitespace-only", i+1)
		}
		if trimmed != seg {
			return fmt.Errorf("segment %d has leading/trailing whitespace; trim or fix the path", i+1)
		}
		if strings.ContainsAny(seg, "[]") {
			return fmt.Errorf("segment %d contains unsupported character (array/index syntax not supported)", i+1)
		}
	}
	return nil
}

func validateConnectorCalloutProjectionBodyField(spec connectorCalloutProfileBatchSpec) error {
	raw, ok := spec.ExtraBodyFields[spec.ProjectionBodyField]
	if !ok {
		return nil
	}
	var existing []string
	if err := json.Unmarshal(raw, &existing); err != nil {
		return fmt.Errorf("extraBodyFields[%q] must be a JSON string array: %w", spec.ProjectionBodyField, err)
	}
	if len(existing) > 0 {
		return fmt.Errorf("extraBodyFields[%q] must be an empty array; field inventory is owned by inbound connector_field_mapping rows", spec.ProjectionBodyField)
	}
	return nil
}

func countExecutableConnectorCalloutObjectPaths(objectPathMap map[string]string) int {
	count := 0
	for objectKind, objectPath := range objectPathMap {
		if strings.TrimSpace(objectKind) == "" || strings.TrimSpace(objectPath) == "" {
			continue
		}
		count++
	}
	return count
}

func (c *connectorCalloutProfileComponent) Kind() ComponentKind {
	return ComponentKindConnectorCalloutProfile
}

func (c *connectorCalloutProfileComponent) PackName() string {
	return c.packName
}

func (c *connectorCalloutProfileComponent) BusinessKey() string {
	return JoinComponentBusinessKey(c.packName, c.definition.ProfileName)
}

func (c *connectorCalloutProfileComponent) Manageability() Manageability {
	return ManageabilityLocked
}

func (c *connectorCalloutProfileComponent) ExpectedPayload() ([]byte, error) {
	return append([]byte(nil), c.payload...), nil
}

func (c *connectorCalloutProfileComponent) ExpectedProjection() (any, error) {
	desc := ""
	if c.definition.Description != nil {
		desc = *c.definition.Description
	}
	return clonePackComponentProjection(connectorCalloutProfileProjectionRow{
		OwnerPack:     c.packName,
		ProfileName:   c.definition.ProfileName,
		CredentialRef: c.definition.CredentialRef,
		EndpointClass: c.definition.EndpointClass,
		RetryPolicy:   c.definition.RetryPolicy,
		PageSpec:      c.definition.PageSpec,
		BatchSpec:     c.definition.BatchSpec,
		ObjectPathMap: c.definition.ObjectPathMap,
		Description:   desc,
	})
}

func (c *connectorCalloutProfileComponent) LiveProjection(ctx context.Context, deps VerifyDeps) (any, error) {
	packInstallID, err := resolveConnectorCalloutProfilePackInstallID(ctx, deps.DB, c.packName)
	if err != nil {
		return nil, err
	}
	return liveConnectorCalloutProfileProjection(ctx, deps.DB, c.packName, packInstallID, c.definition.ProfileName)
}

func (c *connectorCalloutProfileComponent) Apply(ctx context.Context, deps ApplyDeps) error {
	return upsertConnectorCalloutProfileDefinition(ctx, deps.DB, c.packName, c.definition)
}

func (c *connectorCalloutProfileComponent) VerifyComponent(ctx context.Context, deps VerifyDeps) error {
	if deps.DB == nil {
		return fmt.Errorf("database is required")
	}
	packInstallID, err := resolveConnectorCalloutProfilePackInstallID(ctx, deps.DB, c.packName)
	if err != nil {
		return fmt.Errorf("VerifyComponent: %w", err)
	}
	expectedAny, err := c.ExpectedProjection()
	if err != nil {
		return fmt.Errorf("expected projection: %w", err)
	}
	expected, ok := expectedAny.(connectorCalloutProfileProjectionRow)
	if !ok {
		return fmt.Errorf("expected connectorCalloutProfileProjectionRow, got %T", expectedAny)
	}
	liveAny, err := liveConnectorCalloutProfileProjection(ctx, deps.DB, c.packName, packInstallID, c.definition.ProfileName)
	if err != nil {
		return fmt.Errorf("live projection: %w", err)
	}
	live, ok := liveAny.(connectorCalloutProfileProjectionRow)
	if !ok {
		return fmt.Errorf("live connectorCalloutProfileProjectionRow, got %T", liveAny)
	}
	// Use canonical JSON comparison instead of reflect.DeepEqual to avoid
	// false drift on json.RawMessage fields (e.g. BatchSpec.ExtraBodyFields)
	// where manifest-indented bytes differ from DB-compact bytes despite
	// being semantically identical. See canonicalProjectionJSON and #1806.
	expectedCanonical, err := canonicalProjectionJSON(expected)
	if err != nil {
		return fmt.Errorf("canonicalize expected projection: %w", err)
	}
	liveCanonical, err := canonicalProjectionJSON(live)
	if err != nil {
		return fmt.Errorf("canonicalize live projection: %w", err)
	}
	if string(expectedCanonical) == string(liveCanonical) {
		return nil
	}
	return fmt.Errorf("projection drift: expected %s, live %s", expectedCanonical, liveCanonical)
}

// upsertConnectorCalloutProfileDefinition inserts or updates a row in
// _System_ConnectorCalloutProfile.
func upsertConnectorCalloutProfileDefinition(ctx context.Context, db *sql.DB, ownerPack string, def connectorCalloutProfileDefinition) error {
	normalizedOwnerPack := strings.TrimSpace(ownerPack)
	normalizedProfileName := strings.ToLower(strings.TrimSpace(def.ProfileName))
	packInstallID, err := resolveConnectorCalloutProfilePackInstallID(ctx, db, normalizedOwnerPack)
	if err != nil {
		return err
	}

	retryJSON, err := json.Marshal(def.RetryPolicy)
	if err != nil {
		return fmt.Errorf("marshal RetryPolicy: %w", err)
	}
	pageJSON, err := json.Marshal(def.PageSpec)
	if err != nil {
		return fmt.Errorf("marshal PageSpec: %w", err)
	}
	batchJSON, err := json.Marshal(def.BatchSpec)
	if err != nil {
		return fmt.Errorf("marshal BatchSpec: %w", err)
	}
	objMapJSON, err := json.Marshal(def.ObjectPathMap)
	if err != nil {
		return fmt.Errorf("marshal ObjectPathMap: %w", err)
	}

	var descParam interface{}
	if def.Description != nil {
		descParam = *def.Description
	}

	_, err = db.ExecContext(ctx, `
INSERT INTO _System_ConnectorCalloutProfile
  (Id, InstalledPackId, OwnerPack, ProfileName, CredentialRef, EndpointClass, RetryPolicy, PageSpec, BatchSpec, ObjectPathMap, Description, IsDeleted, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, NOW(), NOW())
ON DUPLICATE KEY UPDATE
  InstalledPackId = VALUES(InstalledPackId),
  OwnerPack = VALUES(OwnerPack),
  CredentialRef = VALUES(CredentialRef),
  EndpointClass = VALUES(EndpointClass),
  RetryPolicy = VALUES(RetryPolicy),
  PageSpec = VALUES(PageSpec),
  BatchSpec = VALUES(BatchSpec),
  ObjectPathMap = VALUES(ObjectPathMap),
  Description = VALUES(Description),
  IsDeleted = 0,
  LastModifiedDate = NOW()`,
		connectorCalloutProfileID(packInstallID, normalizedProfileName), packInstallID, normalizedOwnerPack, normalizedProfileName, def.CredentialRef, def.EndpointClass,
		string(retryJSON), string(pageJSON), string(batchJSON), string(objMapJSON), descParam,
	)
	if err != nil {
		return fmt.Errorf("upsert _System_ConnectorCalloutProfile %s/%s: %w", normalizedOwnerPack, normalizedProfileName, err)
	}
	if _, candidate := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, normalizedOwnerPack); candidate {
		// The selected incarnation remains authoritative until the activation
		// transaction. Candidate apply may create its exact snapshot but must not
		// retire the previous snapshot early.
		return nil
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorCalloutProfile
   SET IsDeleted = 1, LastModifiedDate = NOW()
 WHERE OwnerPack = ?
   AND ProfileName = ?
   AND COALESCE(InstalledPackId, '') <> ?
   AND COALESCE(IsDeleted, 0) = 0`,
		normalizedOwnerPack, normalizedProfileName, packInstallID,
	); err != nil {
		return fmt.Errorf("soft-delete stale _System_ConnectorCalloutProfile incarnations %s/%s: %w", normalizedOwnerPack, normalizedProfileName, err)
	}
	return nil
}

func resolveConnectorCalloutProfilePackInstallID(ctx context.Context, db *sql.DB, packName string) (string, error) {
	if packInstallID, ok := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName); ok {
		return packInstallID, nil
	}
	var packInstallID string
	err := db.QueryRowContext(ctx, `
    SELECT PackInstallId
      FROM _System_InstalledPack
     WHERE Name = ?
       AND COALESCE(IsDeleted, 0) = 0
       AND Status IN ('installing', 'active')
     ORDER BY LastModifiedDate DESC
     LIMIT 1`, packName).Scan(&packInstallID)
	if err == sql.ErrNoRows {
		return "", fmt.Errorf("connector_callout_profile: pack %q has no in-flight installation record (Apply called outside install lifecycle?)", packName)
	}
	if err != nil {
		return "", fmt.Errorf("connector_callout_profile: lookup PackInstallId for pack %q: %w", packName, err)
	}
	if strings.TrimSpace(packInstallID) == "" {
		return "", fmt.Errorf("connector_callout_profile: pack %q has empty PackInstallId (upgrade backfill not applied)", packName)
	}
	return packInstallID, nil
}

func connectorCalloutProfileID(packInstallID, profileName string) string {
	sum := sha256.Sum256([]byte(
		strings.ToLower(strings.TrimSpace(packInstallID)) + "\x00" +
			strings.ToLower(strings.TrimSpace(profileName)),
	))
	return "ccp_" + hex.EncodeToString(sum[:])
}

// liveConnectorCalloutProfileProjection reads the live row from
// _System_ConnectorCalloutProfile and returns the projection shape.
// All behavior-defining columns (RetryPolicy, PageSpec, BatchSpec, ObjectPathMap,
// Description) are included so drift detection catches mutations beyond identity fields.
func liveConnectorCalloutProfileProjection(ctx context.Context, db *sql.DB, ownerPack, packInstallID, profileName string) (any, error) {
	var row connectorCalloutProfileProjectionRow
	var retryJSON, pageJSON, batchJSON, objMapJSON string
	err := db.QueryRowContext(ctx, `
SELECT InstalledPackId, OwnerPack, ProfileName, CredentialRef, EndpointClass,
       COALESCE(RetryPolicy, ''), COALESCE(PageSpec, ''),
       COALESCE(BatchSpec, ''), COALESCE(ObjectPathMap, ''),
       COALESCE(Description, '')
FROM _System_ConnectorCalloutProfile
WHERE OwnerPack = ? AND InstalledPackId = ? AND ProfileName = ? AND COALESCE(IsDeleted, 0) = 0
LIMIT 1`, ownerPack, packInstallID, profileName).Scan(
		&row.InstalledPackID, &row.OwnerPack, &row.ProfileName, &row.CredentialRef, &row.EndpointClass,
		&retryJSON, &pageJSON, &batchJSON, &objMapJSON, &row.Description)
	if err == sql.ErrNoRows {
		return nil, fmt.Errorf("connector callout profile %s/%s (%s): row missing or soft-deleted", ownerPack, profileName, packInstallID)
	}
	if err != nil {
		return nil, fmt.Errorf("query connector callout profile %s/%s (%s): %w", ownerPack, profileName, packInstallID, err)
	}
	if retryJSON != "" {
		if err := json.Unmarshal([]byte(retryJSON), &row.RetryPolicy); err != nil {
			return nil, fmt.Errorf("connector callout profile %s/%s (%s): unmarshal RetryPolicy: %w", ownerPack, profileName, packInstallID, err)
		}
	}
	if pageJSON != "" {
		if err := json.Unmarshal([]byte(pageJSON), &row.PageSpec); err != nil {
			return nil, fmt.Errorf("connector callout profile %s/%s (%s): unmarshal PageSpec: %w", ownerPack, profileName, packInstallID, err)
		}
	}
	if batchJSON != "" {
		if err := json.Unmarshal([]byte(batchJSON), &row.BatchSpec); err != nil {
			return nil, fmt.Errorf("connector callout profile %s/%s (%s): unmarshal BatchSpec: %w", ownerPack, profileName, packInstallID, err)
		}
	}
	if objMapJSON != "" {
		if err := json.Unmarshal([]byte(objMapJSON), &row.ObjectPathMap); err != nil {
			return nil, fmt.Errorf("connector callout profile %s/%s (%s): unmarshal ObjectPathMap: %w", ownerPack, profileName, packInstallID, err)
		}
	}
	// ExpectedProjection deliberately omits InstalledPackId — same pattern as
	// connector_inbound_route_component (Agent A Gap 4 fix). Identity is still
	// guaranteed by the SQL WHERE InstalledPackId = ? clause above.
	row.InstalledPackID = ""
	return row, nil
}

// connectorCalloutProfileComponentsPayload is the JSON envelope used in pack manifests.
type connectorCalloutProfileComponentsPayload struct {
	ConnectorCalloutProfiles []connectorCalloutProfileDefinition `json:"connectorCalloutProfiles"`
}

// parseConnectorCalloutProfileComponents parses a JSON payload containing a list of
// callout profile definitions and returns a slice of PackComponent objects.
func parseConnectorCalloutProfileComponents(packName string, payload []byte) ([]PackComponent, error) {
	var data connectorCalloutProfileComponentsPayload
	if err := json.Unmarshal(payload, &data); err != nil {
		return nil, fmt.Errorf("parse connector_callout_profiles payload: %w", err)
	}
	out := make([]PackComponent, 0, len(data.ConnectorCalloutProfiles))
	for _, def := range data.ConnectorCalloutProfiles {
		c, err := NewConnectorCalloutProfileComponent(packName, def)
		if err != nil {
			return nil, fmt.Errorf("connector callout profile component %s/%s: %w", packName, def.ProfileName, err)
		}
		out = append(out, c)
	}
	// Stable sort for deterministic payload hash.
	sort.Slice(out, func(i, j int) bool {
		return out[i].BusinessKey() < out[j].BusinessKey()
	})
	return out, nil
}