package connector_outboundintent

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strconv"
	"strings"

	connector_callout "github.com/pingcap-inc/nexuscrm-connectors/features/connector/callout"
	connectorv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

type calloutEngine interface {
	Do(context.Context, connector_callout.CalloutRequest, connector_callout.RetryPolicy) (connector_callout.CalloutResponse, error)
}

type ProviderPreparer struct {
	db        *sql.DB
	authority *Authority
	callouts  calloutEngine
}

func NewProviderPreparer(db *sql.DB, authority *Authority, callouts calloutEngine) (*ProviderPreparer, error) {
	if db == nil || authority == nil || callouts == nil || authority.Scope().validate() != nil {
		return nil, ErrInvalidInput
	}
	return &ProviderPreparer{db: db, authority: authority, callouts: callouts}, nil
}

type resolvedAssociation struct {
	contract      connectorv1.ConnectorPushAssociationContract
	remoteID      string
	linkSnapshots []preparedRemoteLinkSnapshot
}

type providerEffect struct {
	callouts             calloutEngine
	profile              *connector_callout.ConnectorCalloutProfile
	contract             connectorv1.ConnectorPushDurableMetadataContract
	main                 connector_callout.CalloutRequest
	phase                string
	associationOnly      bool
	checkpointStatus     int
	attemptCount         int
	maxAttempts          int
	remoteID             string
	body                 []byte
	associations         []resolvedAssociation
	linkFinalizationBase RemoteLinkFinalization
	scopedLinkProof      *preparedScopedRemoteLink
	legacyProof          *preparedLegacyRemoteLink
}

func (e *providerEffect) scopedRemoteLinkProof() *preparedScopedRemoteLink {
	return e.scopedLinkProof
}

func (e *providerEffect) legacyRemoteLinkProof() *preparedLegacyRemoteLink {
	return e.legacyProof
}

func checkpointedAssociationPhase(command Command) (string, bool) {
	if strings.TrimSpace(command.RemoteID) == "" || command.ResponseStatus < 200 || command.ResponseStatus >= 300 {
		return "", false
	}
	switch command.FailureCode {
	case "association_create_outcome_uncertain":
		return "create", true
	case "association_update_outcome_uncertain":
		return "update", true
	default:
		return "", false
	}
}

type providerDesiredState struct {
	payload          outboundintentv1.UpsertDesiredStateV1
	writeDisposition outboundintentv1.UpsertWriteDisposition
}

func decodeProviderDesiredState(intent outboundintentv1.Intent) (providerDesiredState, error) {
	switch intent.PayloadFormatVersion {
	case 1:
		payload, err := outboundintentv1.DecodeUpsertDesiredStateV1(intent)
		if err != nil {
			return providerDesiredState{}, err
		}
		return providerDesiredState{
			payload: payload, writeDisposition: outboundintentv1.UpsertWriteCreateOrUpdate,
		}, nil
	case 2:
		payload, err := outboundintentv1.DecodeUpsertDesiredStateV2(intent)
		if err != nil {
			return providerDesiredState{}, err
		}
		return providerDesiredState{
			payload: outboundintentv1.UpsertDesiredStateV1{
				Version: 1, TargetObject: payload.TargetObject, BodyRoot: payload.BodyRoot,
				MappedFields: payload.MappedFields, Associations: payload.Associations,
			},
			writeDisposition: payload.WriteDisposition,
		}, nil
	default:
		return providerDesiredState{}, ErrInvalidInput
	}
}

func selectProviderWritePhase(disposition outboundintentv1.UpsertWriteDisposition, remoteLinkFound bool) (string, error) {
	if !disposition.Valid() {
		return "", ErrInvalidInput
	}
	if remoteLinkFound {
		return "update", nil
	}
	if disposition == outboundintentv1.UpsertWriteUpdateExistingOnly {
		return "", ErrIdentityConflict
	}
	return "create", nil
}

func (p *ProviderPreparer) Prepare(ctx context.Context, command Command) (PreparedEffect, error) {
	if p == nil || p.db == nil || p.authority == nil || p.callouts == nil || ctx == nil {
		return nil, ErrDependency
	}
	desiredState, err := decodeProviderDesiredState(command.Intent)
	if err != nil {
		return nil, &PreparationError{Code: "invalid_desired_state", Cause: err}
	}
	payload := desiredState.payload
	resolved, err := p.authority.Resolve(ctx, p.db, command)
	if err != nil {
		return nil, err
	}
	contract := resolved.Contract
	if payload.TargetObject != contract.TargetObject || payload.BodyRoot != contract.BodyRoot {
		return nil, &PreparationError{Code: "desired_state_mismatch", Cause: ErrIdentityConflict}
	}
	associations, err := p.resolveAssociations(ctx, payload, contract)
	if err != nil {
		return nil, err
	}
	link, found, err := readRemoteLink(ctx, p.db, command.LinkKey)
	if err != nil {
		return nil, &PreparationError{Code: "remote_link_unavailable", Retryable: true, Cause: err}
	}
	if found && !validDispatchRemoteLink(link, command) {
		return nil, &PreparationError{Code: "remote_link_conflict", Cause: ErrIdentityConflict}
	}
	if found {
		link, err = ensureRemoteLinkProfile(ctx, p.db, link, resolvedProfileAuthority{
			Profile: resolved.Profile, Binding: resolved.ProfileBinding,
			RetryPolicyJSON: resolved.RetryPolicyJSON, ProfileHash: resolved.ProfileHash,
		})
		if err != nil {
			return nil, &PreparationError{
				Code: "remote_link_conflict", Retryable: authorityLoadRetryable(err), Cause: err,
			}
		}
		if validateRemoteLinkExecutionSnapshot(link, command, contract, resolved) != nil {
			return nil, &PreparationError{Code: "remote_link_conflict", Cause: ErrIdentityConflict}
		}
	}
	associationLinks := make([]preparedRemoteLinkSnapshot, 0, len(associations))
	for _, association := range associations {
		associationLinks = append(associationLinks, association.linkSnapshots...)
	}
	scopedLinkProof := newPreparedScopedRemoteLink(command, link, found, associationLinks)
	var legacyProof *preparedLegacyRemoteLink
	if !found && desiredState.writeDisposition == outboundintentv1.UpsertWriteUpdateExistingOnly {
		link, found, err = readRemoteLink(ctx, p.db, legacyRemoteLinkKey(command))
		if err != nil {
			return nil, &PreparationError{Code: "remote_link_unavailable", Retryable: true, Cause: err}
		}
		if found {
			legacyProof, err = p.prepareLegacyRemoteLink(ctx, command, link, resolved)
			if err != nil {
				return nil, &PreparationError{Code: "remote_link_conflict", Retryable: authorityLoadRetryable(err), Cause: err}
			}
		}
	}
	phase, err := selectProviderWritePhase(desiredState.writeDisposition, found)
	if err != nil {
		return nil, &PreparationError{Code: "remote_link_required", Cause: err}
	}
	body, err := buildProviderBody(payload, associations, phase, contract.TargetObject, link.RemoteID)
	if err != nil {
		return nil, &PreparationError{Code: "invalid_desired_state", Cause: err}
	}
	request := connector_callout.CalloutRequest{
		Headers:       http.Header{"Content-Type": []string{"application/json"}},
		Body:          body,
		CredentialRef: resolved.Profile.CredentialRef,
		EndpointClass: resolved.Profile.EndpointClass,
	}
	remoteID := ""
	if phase == "update" {
		remoteID = link.RemoteID
		request.Method = http.MethodPatch
		request.RelativePath, err = renderUpdatePath(contract.UpdatePathTemplate, remoteID)
	} else {
		request.Method = http.MethodPost
		request.RelativePath, err = concreteCreatePath(contract.CreatePathTemplate)
		request.Headers.Set("Idempotency-Key", command.Intent.IdempotencyKey)
	}
	if err != nil {
		return nil, &PreparationError{Code: "invalid_execution_binding", Cause: err}
	}
	finalization := RemoteLinkFinalization{
		Create: phase == "create", Scope: command.Scope,
		LinkKey: command.LinkKey, RemoteID: remoteID,
		ConnectorOwnerKey: contract.ConnectorOwnerKey, ActionSlotID: contract.ActionSlotID,
		SourceObject: contract.SourceObject, SourceRecordID: command.Intent.SourceRecordID,
		DestinationSlotID:       contract.DestinationSlotID,
		RequesterOwnerPack:      contract.ConnectorOwnerKey,
		ProfileOwnerPack:        resolved.Profile.OwnerPack,
		FlowDeveloperName:       contract.FlowDeveloperName,
		FlowActionName:          contract.FlowActionName,
		TargetObject:            contract.TargetObject,
		ProfileName:             resolved.Profile.ProfileName,
		PackInstallationID:      resolved.Profile.InstalledPackID,
		CredentialRef:           resolved.Profile.CredentialRef,
		EndpointClass:           resolved.Profile.EndpointClass,
		RetryPolicyJSON:         resolved.RetryPolicyJSON,
		ProfileContractHash:     resolved.ProfileHash,
		CreatePathTemplate:      contract.CreatePathTemplate,
		UpdatePathTemplate:      contract.UpdatePathTemplate,
		ResponseIDPath:          contract.ResponseIDPath,
		RemoteIDField:           contract.RemoteIDField,
		RemoteIDProvenanceField: contract.RemoteIDProvenanceField,
		MappingFingerprint:      resolved.MappingFingerprint,
		BodyFingerprint:         resolved.BodyFingerprint,
	}
	if associationPhase, checkpointed := checkpointedAssociationPhase(command); checkpointed {
		if !found || link.RemoteID != command.RemoteID {
			return nil, &PreparationError{Code: "remote_link_conflict", Cause: ErrIdentityConflict}
		}
		return &providerEffect{
			callouts: p.callouts, profile: resolved.Profile, contract: contract,
			phase: associationPhase, associationOnly: true, checkpointStatus: command.ResponseStatus,
			attemptCount: command.AttemptCount + 1, maxAttempts: normalizedMaxAttempts(resolved.Profile.RetryPolicy.MaxAttempts),
			remoteID: link.RemoteID, body: body, associations: associations,
			linkFinalizationBase: finalization, scopedLinkProof: scopedLinkProof, legacyProof: legacyProof,
		}, nil
	}
	return &providerEffect{
		callouts: p.callouts, profile: resolved.Profile, contract: contract,
		main: request, phase: phase, remoteID: remoteID, body: body,
		attemptCount: command.AttemptCount + 1, maxAttempts: normalizedMaxAttempts(resolved.Profile.RetryPolicy.MaxAttempts),
		associations: associations, linkFinalizationBase: finalization,
		scopedLinkProof: scopedLinkProof, legacyProof: legacyProof,
	}, nil
}

func normalizedMaxAttempts(value int) int {
	if value < 1 {
		return 1
	}
	return value
}

func (p *ProviderPreparer) resolveAssociations(
	ctx context.Context,
	payload outboundintentv1.UpsertDesiredStateV1,
	contract connectorv1.ConnectorPushDurableMetadataContract,
) ([]resolvedAssociation, error) {
	result := make([]resolvedAssociation, 0, len(payload.Associations))
	for _, source := range payload.Associations {
		if int(source.ContractIndex) >= len(contract.Associations) {
			return nil, &PreparationError{Code: "association_mismatch", Cause: ErrIdentityConflict}
		}
		binding := contract.Associations[source.ContractIndex]
		if binding.Junction != nil || binding.SourceObject != source.SourceObject {
			return nil, &PreparationError{Code: "association_mismatch", Cause: ErrIdentityConflict}
		}
		targetProfile, err := p.authority.resolveNamedProfile(
			ctx, p.db, binding.TargetProfileOwnerPack, binding.TargetProfileName,
		)
		if err != nil {
			return nil, &PreparationError{
				Code: "association_profile_unavailable", Retryable: authorityLoadRetryable(err), Cause: err,
			}
		}
		actionSlotIDs := associationTargetActionSlotIDs(binding)
		var link remoteLinkIdentity
		found := false
		linkSnapshots := make([]preparedRemoteLinkSnapshot, 0, len(actionSlotIDs))
		for _, actionSlotID := range actionSlotIDs {
			linkKey := linkKeyForIdentity(
				p.authority.Scope(), contract.ConnectorOwnerKey, actionSlotID,
				binding.SourceObject, source.SourceRecordID, binding.TargetDestinationSlotID,
			)
			link, found, err = readRemoteLink(ctx, p.db, linkKey)
			if err != nil {
				return nil, &PreparationError{Code: "association_link_unavailable", Retryable: true, Cause: err}
			}
			snapshot := preparedRemoteLinkSnapshot{linkKey: linkKey, found: found}
			if found {
				snapshot.link = link
			}
			linkSnapshots = append(linkSnapshots, snapshot)
			if found {
				break
			}
		}
		if !found {
			if len(linkSnapshots) == 0 {
				return nil, &PreparationError{Code: "association_mismatch", Cause: ErrIdentityConflict}
			}
			readinessKeys := make([]string, len(linkSnapshots))
			for index := range linkSnapshots {
				readinessKeys[index] = linkSnapshots[index].linkKey
			}
			return nil, &dependencyPreparationError{
				Code: "association_link_pending",
				Dependency: associationRemoteLinkDependency(
					p.authority.Scope(), contract.ConnectorOwnerKey, binding.SourceObject,
					source.SourceRecordID, binding.TargetDestinationSlotID, readinessKeys,
				),
				Cause: ErrDependency,
			}
		}
		if validateAssociationRemoteLinkIdentity(
			link, contract.ConnectorOwnerKey, binding, source.SourceRecordID, actionSlotIDs,
		) != nil {
			return nil, &PreparationError{Code: "association_link_conflict", Cause: ErrIdentityConflict}
		}
		link, err = ensureRemoteLinkProfile(ctx, p.db, link, targetProfile)
		if err != nil {
			return nil, &PreparationError{
				Code: "association_link_conflict", Retryable: authorityLoadRetryable(err), Cause: err,
			}
		}
		if validateAssociationRemoteLinkIdentity(
			link, contract.ConnectorOwnerKey, binding, source.SourceRecordID, actionSlotIDs,
		) != nil {
			return nil, &PreparationError{Code: "association_link_conflict", Cause: ErrIdentityConflict}
		}
		if !validRemoteLinkAuthority(p.authority.Scope(), link) {
			return nil, &PreparationError{Code: "association_link_conflict", Cause: ErrIdentityConflict}
		}
		// ensureRemoteLinkProfile can durably refresh the selected link. Keep
		// the negative prefix decisions, but fence dispatch against the exact
		// post-refresh identity that was used to build the provider request.
		linkSnapshots[len(linkSnapshots)-1].link = link
		result = append(result, resolvedAssociation{
			contract: binding, remoteID: link.RemoteID, linkSnapshots: linkSnapshots,
		})
	}
	return result, nil
}

func (e *providerEffect) Dispatch(ctx context.Context) EffectResult {
	return e.dispatch(ctx, nil)
}

func (e *providerEffect) DispatchWithProviderSuccessCheckpoint(
	ctx context.Context,
	checkpoint ProviderSuccessCheckpoint,
) EffectResult {
	return e.dispatch(ctx, checkpoint)
}

func (e *providerEffect) dispatch(
	ctx context.Context,
	checkpoint ProviderSuccessCheckpoint,
) EffectResult {
	if e == nil || e.callouts == nil || e.profile == nil || ctx == nil {
		return EffectResult{Outcome: EffectUncertain, FailureCode: "invalid_effect"}
	}
	if e.associationOnly {
		if result := e.dispatchAssociations(ctx, e.remoteID); result != nil {
			return e.retryAfterMainSuccess(*result, e.remoteID, e.checkpointStatus, e.attemptCount)
		}
		finalization := e.linkFinalizationBase
		finalization.Create = false
		finalization.RemoteID = e.remoteID
		return EffectResult{
			Outcome: EffectSucceeded, ResponseStatus: e.checkpointStatus,
			RemoteID: e.remoteID, LinkFinalization: &finalization,
		}
	}
	policy := e.profile.RetryPolicy
	// Integration's durable command owner, rather than the in-memory HTTP
	// helper, owns every provider mutation retry. A lost POST or PATCH response
	// may already represent a committed remote effect.
	policy.MaxAttempts = 1
	response, callErr := e.callouts.Do(ctx, e.main, policy)
	if e.phase == "create" && e.matchesCreateConflict(response) {
		return e.resolveCreateConflict(ctx, response)
	}
	if result, terminal := classifyProviderResult(response, callErr); terminal {
		if result.Outcome == EffectNotDispatched && e.attemptCount >= normalizedMaxAttempts(e.maxAttempts) {
			result.Outcome, result.FailureCode = EffectRejectedNoEffect, "provider_retry_exhausted"
		}
		return result
	}
	remoteID := e.remoteID
	if e.phase == "create" {
		var err error
		remoteID, err = extractResponseID(response.Body, e.contract.ResponseIDPath)
		if err != nil {
			return EffectResult{Outcome: EffectUncertain, ResponseStatus: response.StatusCode, FailureCode: "response_identity_missing"}
		}
	}
	if checkpoint != nil && e.hasAssociationForPhase() {
		finalization := e.linkFinalizationBase
		finalization.RemoteID = remoteID
		code := "association_" + e.phase + "_outcome_uncertain"
		if err := checkpoint(ctx, response.StatusCode, finalization, code); err != nil {
			return EffectResult{
				Outcome: EffectUncertain, ResponseStatus: response.StatusCode,
				RemoteID: remoteID, FailureCode: "provider_success_checkpoint_failed",
			}
		}
	}
	if result := e.dispatchAssociations(ctx, remoteID); result != nil {
		// The durable provider-success checkpoint starts a new, independently
		// bounded association lifecycle. Earlier main-provider attempts must not
		// consume its retry budget.
		return e.retryAfterMainSuccess(*result, remoteID, response.StatusCode, 1)
	}
	finalization := e.linkFinalizationBase
	finalization.RemoteID = remoteID
	return EffectResult{
		Outcome: EffectSucceeded, ResponseStatus: response.StatusCode, RemoteID: remoteID,
		LinkFinalization: &finalization,
	}
}

func (e *providerEffect) hasAssociationForPhase() bool {
	for _, association := range e.associations {
		if phaseEnabled(association.contract.CalloutPhases, e.phase) {
			return true
		}
	}
	return false
}

func (e *providerEffect) retryAfterMainSuccess(
	failure EffectResult,
	remoteID string,
	mainStatus int,
	associationAttempt int,
) EffectResult {
	finalization := e.linkFinalizationBase
	finalization.RemoteID = remoteID
	if failure.Outcome == EffectRejected {
		return EffectResult{
			Outcome: EffectTerminalAfterSuccess, ResponseStatus: mainStatus,
			RemoteID: remoteID, FailureCode: "association_" + e.phase + "_rejected",
			LinkFinalization: &finalization,
		}
	}
	if associationAttempt >= normalizedMaxAttempts(e.maxAttempts) {
		return EffectResult{
			Outcome: EffectTerminalAfterSuccess, ResponseStatus: mainStatus,
			RemoteID: remoteID, FailureCode: "association_" + e.phase + "_retry_exhausted",
			LinkFinalization: &finalization,
		}
	}
	return EffectResult{
		Outcome: EffectRetryableAfterSuccess, ResponseStatus: mainStatus,
		RemoteID: remoteID, FailureCode: "association_" + e.phase + "_outcome_uncertain",
		RetryAfter: failure.RetryAfter, LinkFinalization: &finalization,
	}
}

func classifyProviderResult(response connector_callout.CalloutResponse, callErr error) (EffectResult, bool) {
	if response.StatusCode >= 200 && response.StatusCode < 300 {
		return EffectResult{}, false
	}
	if response.StatusCode != 0 {
		if response.StatusCode == http.StatusTooManyRequests || response.StatusCode >= 500 {
			return EffectResult{Outcome: EffectUncertain, ResponseStatus: response.StatusCode, FailureCode: "provider_outcome_uncertain"}, true
		}
		return EffectResult{Outcome: EffectRejected, ResponseStatus: response.StatusCode, FailureCode: "provider_status_" + strconv.Itoa(response.StatusCode)}, true
	}
	for _, candidate := range []struct {
		err       error
		code      string
		retryable bool
	}{
		{connector_callout.ErrEndpointUnavailable, "endpoint_unavailable", true},
		{connector_callout.ErrCredentialUnavailable, "credential_unavailable", true},
		{connector_callout.ErrInvalidRequest, "invalid_request", false},
		{connector_callout.ErrCredentialedEndpointMustBeHTTPS, "insecure_endpoint", false},
		{connector_callout.ErrUnsupportedAuthType, "unsupported_auth", false},
	} {
		if errors.Is(callErr, candidate.err) {
			if candidate.retryable {
				return EffectResult{Outcome: EffectNotDispatched, FailureCode: candidate.code}, true
			}
			return EffectResult{Outcome: EffectRejectedNoEffect, FailureCode: candidate.code}, true
		}
	}
	return EffectResult{Outcome: EffectUncertain, FailureCode: "provider_outcome_uncertain"}, true
}

func (e *providerEffect) dispatchAssociations(ctx context.Context, fromID string) *EffectResult {
	for _, association := range e.associations {
		if !phaseEnabled(association.contract.CalloutPhases, e.phase) {
			continue
		}
		method := strings.ToUpper(strings.TrimSpace(association.contract.AssociationMethod))
		if method != http.MethodPut {
			return &EffectResult{Outcome: EffectRejected, FailureCode: "association_not_idempotent"}
		}
		path, pathErr := renderAssociationPath(
			association.contract.AssociationPath, e.contract.TargetObject, fromID,
			association.contract.TargetObject, association.remoteID,
		)
		if pathErr != nil {
			return &EffectResult{Outcome: EffectRejected, FailureCode: "association_binding_invalid"}
		}
		var body []byte
		if association.contract.CalloutBodyTemplate != nil {
			rendered := renderAssociationTemplate(
				association.contract.CalloutBodyTemplate, e.contract.TargetObject,
				fromID, association.contract.TargetObject, association.remoteID,
			)
			var marshalErr error
			body, marshalErr = json.Marshal(rendered)
			if marshalErr != nil {
				return &EffectResult{Outcome: EffectRejected, FailureCode: "association_binding_invalid"}
			}
		}
		response, err := e.callouts.Do(ctx, connector_callout.CalloutRequest{
			Method: method, RelativePath: path, Body: body,
			Headers:       http.Header{"Content-Type": []string{"application/json"}},
			CredentialRef: e.profile.CredentialRef, EndpointClass: e.profile.EndpointClass,
		}, connector_callout.RetryPolicy{MaxAttempts: 1})
		if response.StatusCode >= 200 && response.StatusCode < 300 && err == nil {
			continue
		}
		if response.StatusCode >= 400 && response.StatusCode < 500 && response.StatusCode != http.StatusTooManyRequests {
			return &EffectResult{Outcome: EffectRejected, ResponseStatus: response.StatusCode, FailureCode: "association_provider_rejected"}
		}
		return &EffectResult{Outcome: EffectNotDispatched, ResponseStatus: response.StatusCode, FailureCode: "association_outcome_uncertain"}
	}
	return nil
}

func (e *providerEffect) matchesCreateConflict(response connector_callout.CalloutResponse) bool {
	policy := e.contract.CreateConflictResolution
	if policy == nil {
		return false
	}
	matchedStatus := false
	for _, status := range policy.StatusCodes {
		matchedStatus = matchedStatus || status == response.StatusCode
	}
	return matchedStatus && diagnosticMatches(response.Body, policy.DiagnosticMatch)
}

func (e *providerEffect) resolveCreateConflict(ctx context.Context, post connector_callout.CalloutResponse) EffectResult {
	policy := e.contract.CreateConflictResolution
	if policy == nil || policy.OnResolved != connectorv1.ConnectorPushConflictOnResolvedAdoptLinkThenPatch {
		return EffectResult{Outcome: EffectRejected, ResponseStatus: post.StatusCode, FailureCode: "provider_conflict"}
	}
	identity, err := scalarAtJSONPath(e.body, policy.Identity.RequestBodyPath)
	if err != nil {
		return EffectResult{Outcome: EffectRejected, ResponseStatus: post.StatusCode, FailureCode: "conflict_identity_missing"}
	}
	if policy.Identity.Normalization == connectorv1.ConnectorPushConflictNormalizationEmail {
		identity = strings.ToLower(strings.TrimSpace(identity))
	}
	path, query, err := renderConflictResolvePath(policy.Resolve, identity)
	if err != nil {
		return EffectResult{Outcome: EffectRejected, ResponseStatus: post.StatusCode, FailureCode: "conflict_resolver_invalid"}
	}
	resolved, resolveErr := e.callouts.Do(ctx, connector_callout.CalloutRequest{
		Method: http.MethodGet, RelativePath: path, QueryParams: query,
		Headers:       http.Header{"Accept": []string{"application/json"}},
		CredentialRef: e.profile.CredentialRef, EndpointClass: e.profile.EndpointClass,
	}, connector_callout.RetryPolicy{MaxAttempts: 1})
	if resolved.StatusCode < 200 || resolved.StatusCode >= 300 || resolveErr != nil {
		return EffectResult{Outcome: EffectUncertain, ResponseStatus: resolved.StatusCode, FailureCode: "conflict_resolution_uncertain"}
	}
	remoteID, err := extractResponseID(resolved.Body, policy.Resolve.ResponseIDPath)
	if err != nil {
		return EffectResult{Outcome: EffectUncertain, ResponseStatus: resolved.StatusCode, FailureCode: "response_identity_missing"}
	}
	if len(e.associations) != 0 {
		return EffectResult{Outcome: EffectRejected, ResponseStatus: resolved.StatusCode, FailureCode: "conflict_association_not_authorized"}
	}
	if policy.NoOldSnapshotPolicy == connectorv1.ConnectorPushConflictNoOldSnapshotAdoptLinkOnly {
		finalization := e.linkFinalizationBase
		finalization.RemoteID = remoteID
		return EffectResult{
			Outcome: EffectSucceeded, ResponseStatus: resolved.StatusCode, RemoteID: remoteID,
			LinkFinalization: &finalization,
		}
	}
	if policy.NoOldSnapshotPolicy != connectorv1.ConnectorPushConflictNoOldSnapshotAdoptLinkThenPatchAuthorizedFrozenCreatePayload {
		return EffectResult{Outcome: EffectRejected, ResponseStatus: resolved.StatusCode, FailureCode: "conflict_patch_not_authorized"}
	}
	finalization := e.linkFinalizationBase
	finalization.RemoteID = remoteID
	return EffectResult{
		Outcome: EffectRetryableAfterSuccess, ResponseStatus: resolved.StatusCode,
		RemoteID: remoteID, FailureCode: "conflict_adopted_patch_pending",
		LinkFinalization: &finalization,
	}
}

func buildProviderBody(payload outboundintentv1.UpsertDesiredStateV1, associations []resolvedAssociation, phase, fromObject, fromID string) ([]byte, error) {
	mapped := make(map[string]any, len(payload.MappedFields))
	for key, raw := range payload.MappedFields {
		decoder := json.NewDecoder(bytes.NewReader(raw))
		decoder.UseNumber()
		var value any
		if err := decoder.Decode(&value); err != nil {
			return nil, err
		}
		mapped[key] = value
	}
	root := map[string]any{}
	if payload.BodyRoot == "" {
		for key, value := range mapped {
			root[key] = value
		}
	} else {
		root[payload.BodyRoot] = mapped
	}
	for _, association := range associations {
		contract := association.contract
		if !phaseEnabled(contract.InlinePhases, phase) || strings.TrimSpace(contract.PayloadField) == "" {
			continue
		}
		rendered := renderAssociationTemplate(contract.PayloadTemplate, fromObject, fromID, contract.TargetObject, association.remoteID)
		switch contract.PayloadMergeMode {
		case "append":
			current, _ := root[contract.PayloadField].([]any)
			root[contract.PayloadField] = append(current, rendered)
		case "set":
			root[contract.PayloadField] = rendered
		default:
			return nil, fmt.Errorf("unsupported association payload merge mode")
		}
	}
	return json.Marshal(root)
}

func readRemoteLink(ctx context.Context, db *sql.DB, linkKey string) (remoteLinkIdentity, bool, error) {
	var link remoteLinkIdentity
	err := db.QueryRowContext(ctx, `
SELECT Id, LinkKey, Status, RemoteId, ConnectorOwnerKey, ActionSlotId,
	   COALESCE(RemoteLocatorHash, ''), COALESCE(RemoteLocatorActiveKey, ''),
	   COALESCE(ConnectorNamespace, ''),
       ActionType, SourceObject, SourceRecordId, DestinationSlotId,
       RequesterOwnerPack, ProfileOwnerPack, FlowDeveloperName, FlowActionName,
       TargetObject, ProfileName, CreatedPackInstallId, LastValidatedPackInstallId,
       CredentialRef, EndpointClass, RetryPolicy, ProfileContractHash,
       CreatePathTemplate, UpdatePathTemplate, ResponseIdPath, RemoteIdField,
       COALESCE(RemoteIdProvenanceField, ''), MappingFingerprint, BodyFingerprint
  FROM _System_ConnectorRemoteLink WHERE LinkKey = ? LIMIT 1`, linkKey).Scan(
		&link.ID, &link.LinkKey, &link.Status, &link.RemoteID,
		&link.ConnectorOwnerKey, &link.ActionSlotID,
		&link.RemoteLocatorHash, &link.RemoteLocatorActiveKey, &link.ConnectorNamespace,
		&link.ActionType,
		&link.SourceObject, &link.SourceRecordID, &link.DestinationSlotID,
		&link.RequesterOwnerPack, &link.ProfileOwnerPack,
		&link.FlowDeveloperName, &link.FlowActionName,
		&link.TargetObject, &link.ProfileName, &link.CreatedPackInstallationID,
		&link.LastValidatedPackInstallationID, &link.CredentialRef,
		&link.EndpointClass, &link.RetryPolicyJSON, &link.ProfileContractHash,
		&link.CreatePathTemplate, &link.UpdatePathTemplate, &link.ResponseIDPath,
		&link.RemoteIDField, &link.RemoteIDProvenanceField,
		&link.MappingFingerprint, &link.BodyFingerprint,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return remoteLinkIdentity{}, false, nil
	}
	return link, err == nil, err
}

func associationTargetActionSlotIDs(contract connectorv1.ConnectorPushAssociationContract) []string {
	values := make([]string, 0, 1+len(contract.TargetPreviousActionSlotIDs))
	seen := make(map[string]struct{}, cap(values))
	candidates := append([]string{contract.TargetActionSlotID}, contract.TargetPreviousActionSlotIDs...)
	for _, candidate := range candidates {
		candidate = strings.TrimSpace(candidate)
		if candidate == "" {
			continue
		}
		if _, duplicate := seen[candidate]; duplicate {
			continue
		}
		seen[candidate] = struct{}{}
		values = append(values, candidate)
	}
	return values
}

func validateRemoteLinkExecutionSnapshot(
	link remoteLinkIdentity,
	command Command,
	contract connectorv1.ConnectorPushDurableMetadataContract,
	resolved ResolvedAuthority,
) error {
	if !sameRemoteLinkForCommand(link, command) || resolved.Profile == nil ||
		link.Status != remoteLinkStatusFinalized || strings.TrimSpace(link.RemoteID) == "" ||
		link.RequesterOwnerPack != contract.ConnectorOwnerKey ||
		link.ProfileOwnerPack != resolved.Profile.OwnerPack ||
		link.FlowDeveloperName != contract.FlowDeveloperName ||
		link.FlowActionName != contract.FlowActionName ||
		link.TargetObject != contract.TargetObject ||
		link.ProfileName != resolved.Profile.ProfileName ||
		link.CredentialRef != resolved.Profile.CredentialRef ||
		link.EndpointClass != resolved.Profile.EndpointClass ||
		link.RetryPolicyJSON != resolved.RetryPolicyJSON ||
		link.ProfileContractHash != resolved.ProfileHash ||
		link.CreatePathTemplate != contract.CreatePathTemplate ||
		link.UpdatePathTemplate != contract.UpdatePathTemplate ||
		link.ResponseIDPath != contract.ResponseIDPath ||
		link.RemoteIDField != contract.RemoteIDField ||
		link.RemoteIDProvenanceField != contract.RemoteIDProvenanceField ||
		link.MappingFingerprint != resolved.MappingFingerprint ||
		link.BodyFingerprint != resolved.BodyFingerprint {
		return ErrIdentityConflict
	}
	return nil
}

func validateAssociationRemoteLinkIdentity(
	link remoteLinkIdentity,
	requesterOwnerPack string,
	binding connectorv1.ConnectorPushAssociationContract,
	sourceRecordID string,
	actionSlotIDs []string,
) error {
	if link.Status != remoteLinkStatusFinalized ||
		link.ConnectorOwnerKey != requesterOwnerPack ||
		!containsString(actionSlotIDs, link.ActionSlotID) ||
		link.ActionType != connectorPushActionType ||
		link.SourceObject != binding.SourceObject ||
		link.SourceRecordID != sourceRecordID ||
		link.DestinationSlotID != binding.TargetDestinationSlotID ||
		strings.TrimSpace(link.RemoteID) == "" ||
		link.RequesterOwnerPack != requesterOwnerPack ||
		link.ProfileOwnerPack != binding.TargetProfileOwnerPack ||
		link.ProfileName != binding.TargetProfileName ||
		link.TargetObject != binding.TargetObject ||
		link.RemoteIDField != binding.RemoteIDField {
		return ErrIdentityConflict
	}
	return nil
}

func ensureRemoteLinkProfile(
	ctx context.Context,
	db *sql.DB,
	link remoteLinkIdentity,
	current resolvedProfileAuthority,
) (remoteLinkIdentity, error) {
	if ctx == nil || db == nil || current.Profile == nil ||
		link.ProfileOwnerPack != current.Profile.OwnerPack ||
		link.ProfileName != current.Profile.ProfileName ||
		link.CredentialRef != current.Profile.CredentialRef ||
		link.EndpointClass != current.Profile.EndpointClass ||
		link.RetryPolicyJSON != current.RetryPolicyJSON ||
		strings.TrimSpace(link.ProfileContractHash) == "" {
		return remoteLinkIdentity{}, ErrIdentityConflict
	}
	if link.ProfileContractHash == current.ProfileHash {
		if link.LastValidatedPackInstallationID != current.Profile.InstalledPackID {
			return remoteLinkIdentity{}, ErrIdentityConflict
		}
		return link, nil
	}
	storedInstall := strings.TrimSpace(link.LastValidatedPackInstallationID)
	if storedInstall == "" {
		storedInstall = strings.TrimSpace(link.CreatedPackInstallationID)
	}
	if storedInstall == "" || link.ProfileContractHash != digestStrings(
		current.Profile.OwnerPack, current.Profile.ProfileName, storedInstall,
		current.Profile.CredentialRef, current.Profile.EndpointClass, current.RetryPolicyJSON,
	) {
		return remoteLinkIdentity{}, ErrIdentityConflict
	}
	result, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
   SET LastValidatedPackInstallId = ?, ProfileContractHash = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
 WHERE LinkKey = ? AND Status = ? AND ProfileContractHash = ?
   AND LastValidatedPackInstallId = ?`,
		current.Profile.InstalledPackID, current.ProfileHash, link.LinkKey,
		remoteLinkStatusFinalized, link.ProfileContractHash, link.LastValidatedPackInstallationID,
	)
	if err != nil {
		return remoteLinkIdentity{}, err
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return remoteLinkIdentity{}, err
	}
	if affected != 1 {
		return remoteLinkIdentity{}, ErrIdentityConflict
	}
	link.LastValidatedPackInstallationID = current.Profile.InstalledPackID
	link.ProfileContractHash = current.ProfileHash
	return link, nil
}

func sameRemoteLinkForCommand(link remoteLinkIdentity, command Command) bool {
	intent := command.Intent
	return link.LinkKey == command.LinkKey && link.ConnectorOwnerKey == intent.ConnectorOwnerKey &&
		link.ActionSlotID == intent.ActionSlotID && link.ActionType == connectorPushActionType &&
		link.SourceObject == intent.SourceObject && link.SourceRecordID == intent.SourceRecordID &&
		link.DestinationSlotID == intent.DestinationSlotID
}

func validDispatchRemoteLink(link remoteLinkIdentity, command Command) bool {
	return sameRemoteLinkForCommand(link, command) &&
		validRemoteLinkAuthority(command.Scope, link)
}

func validRemoteLinkAuthority(scope Scope, link remoteLinkIdentity) bool {
	if scope.validate() != nil || link.Status != remoteLinkStatusFinalized ||
		link.RemoteID == "" || link.RemoteID != strings.TrimSpace(link.RemoteID) {
		return false
	}
	expectedLocator := remoteLocatorHash(
		scope, link.ConnectorOwnerKey, link.DestinationSlotID, link.RemoteID,
	)
	return link.ConnectorNamespace == link.ConnectorOwnerKey &&
		link.RemoteLocatorHash == expectedLocator &&
		link.RemoteLocatorActiveKey == expectedLocator
}

func concreteCreatePath(template string) (string, error) {
	value := strings.TrimSpace(template)
	if value == "" || strings.ContainsAny(value, "{}?#") {
		return "", ErrIdentityConflict
	}
	if !strings.HasPrefix(value, "/") {
		value = "/" + value
	}
	return value, nil
}

func renderUpdatePath(template, remoteID string) (string, error) {
	template = strings.TrimSpace(template)
	remoteID = strings.TrimSpace(remoteID)
	if remoteID == "" || !strings.Contains(template, "{remoteId}") {
		return "", ErrIdentityConflict
	}
	value := strings.ReplaceAll(template, "{remoteId}", url.PathEscape(remoteID))
	if value == "" || strings.ContainsAny(value, "{}?#") {
		return "", ErrIdentityConflict
	}
	if !strings.HasPrefix(value, "/") {
		value = "/" + value
	}
	return value, nil
}

func extractResponseID(body []byte, path string) (string, error) {
	return scalarAtJSONPath(body, path)
}

func scalarAtJSONPath(body []byte, path string) (string, error) {
	var current any
	decoder := json.NewDecoder(bytes.NewReader(body))
	decoder.UseNumber()
	if err := decoder.Decode(&current); err != nil {
		return "", err
	}
	var trailing any
	if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
		if err == nil {
			return "", ErrIdentityConflict
		}
		return "", err
	}
	for _, segment := range strings.Split(strings.Trim(strings.TrimSpace(path), "."), ".") {
		object, ok := current.(map[string]any)
		if !ok || strings.TrimSpace(segment) == "" {
			return "", ErrIdentityConflict
		}
		current, ok = object[segment]
		if !ok {
			return "", ErrIdentityConflict
		}
	}
	var value string
	switch typed := current.(type) {
	case string:
		value = strings.TrimSpace(typed)
	case json.Number:
		value = strings.TrimSpace(typed.String())
	default:
		return "", ErrIdentityConflict
	}
	if value == "" || value == "<nil>" {
		return "", ErrIdentityConflict
	}
	return value, nil
}

func diagnosticMatches(body []byte, expected map[string]string) bool {
	for path, want := range expected {
		value, err := scalarAtJSONPath(body, path)
		if err != nil || value != strings.TrimSpace(want) {
			return false
		}
	}
	return true
}

func renderConflictResolvePath(resolve connectorv1.ConnectorPushConflictResolverContract, identity string) (string, url.Values, error) {
	if strings.ToUpper(strings.TrimSpace(resolve.Method)) != http.MethodGet || !resolve.RequireExactlyOne ||
		!strings.Contains(resolve.PathTemplate, "{identity}") {
		return "", nil, ErrIdentityConflict
	}
	rendered := renderConflictIdentity(resolve.PathTemplate, identity)
	parsed, err := url.Parse(rendered)
	if err != nil || parsed.IsAbs() || parsed.Host != "" || parsed.User != nil || parsed.Fragment != "" || strings.ContainsAny(parsed.Path, "{}") {
		return "", nil, ErrIdentityConflict
	}
	query := parsed.Query()
	path := parsed.EscapedPath()
	if !strings.HasPrefix(path, "/") {
		path = "/" + path
	}
	return path, query, nil
}

func renderConflictIdentity(template, identity string) string {
	var result strings.Builder
	for offset := 0; ; {
		index := strings.Index(template[offset:], "{identity}")
		if index < 0 {
			result.WriteString(template[offset:])
			return result.String()
		}
		index += offset
		result.WriteString(template[offset:index])
		before := template[:index]
		if question := strings.LastIndex(before, "?"); question >= 0 && question > strings.LastIndex(before, "#") {
			result.WriteString(url.QueryEscape(identity))
		} else {
			result.WriteString(url.PathEscape(identity))
		}
		offset = index + len("{identity}")
	}
}

func phaseEnabled(values []string, phase string) bool {
	for _, value := range values {
		if strings.TrimSpace(value) == phase {
			return true
		}
	}
	return false
}

func renderAssociationPath(template, fromObject, fromID, toObject, toID string) (string, error) {
	template = strings.TrimSpace(template)
	fromID = strings.TrimSpace(fromID)
	toID = strings.TrimSpace(toID)
	if fromID == "" || toID == "" ||
		!strings.Contains(template, "{fromId}") || !strings.Contains(template, "{toId}") {
		return "", ErrIdentityConflict
	}
	replacements := map[string]string{
		"{fromObject}": fromObject, "{fromId}": fromID,
		"{toObject}": toObject, "{toId}": toID,
	}
	result := template
	for token, value := range replacements {
		result = strings.ReplaceAll(result, token, url.PathEscape(value))
	}
	result = strings.TrimSpace(result)
	if result == "" || strings.ContainsAny(result, "{}?#") {
		return "", ErrIdentityConflict
	}
	if !strings.HasPrefix(result, "/") {
		result = "/" + result
	}
	return result, nil
}

func renderAssociationTemplate(template any, fromObject, fromID, toObject, toID string) any {
	switch value := template.(type) {
	case map[string]any:
		result := make(map[string]any, len(value))
		for key, item := range value {
			result[key] = renderAssociationTemplate(item, fromObject, fromID, toObject, toID)
		}
		return result
	case []any:
		result := make([]any, len(value))
		for index, item := range value {
			result[index] = renderAssociationTemplate(item, fromObject, fromID, toObject, toID)
		}
		return result
	case string:
		result := value
		for token, replacement := range map[string]string{
			"{fromObject}": fromObject, "{fromId}": fromID,
			"{toObject}": toObject, "{toId}": toID,
		} {
			result = strings.ReplaceAll(result, token, replacement)
		}
		return result
	default:
		return value
	}
}