package connector_outboundintent

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"sort"
	"strings"

	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

const (
	remoteLinkStatusFinalized = "finalized"
	remoteLinkMirrorPending   = "pending"
	connectorPushActionType   = "PushToConnector"
)

type remoteLinkIdentity struct {
	ID, LinkKey, Status, RemoteID   string
	RemoteLocatorHash               string
	RemoteLocatorActiveKey          string
	ConnectorOwnerKey               string
	ConnectorNamespace              string
	ActionSlotID                    string
	ActionType                      string
	SourceObject                    string
	SourceRecordID                  string
	DestinationSlotID               string
	RequesterOwnerPack              string
	ProfileOwnerPack                string
	FlowDeveloperName               string
	FlowActionName                  string
	TargetObject                    string
	ProfileName                     string
	CreatedPackInstallationID       string
	LastValidatedPackInstallationID string
	CredentialRef                   string
	EndpointClass                   string
	RetryPolicyJSON                 string
	ProfileContractHash             string
	CreatePathTemplate              string
	UpdatePathTemplate              string
	ResponseIDPath                  string
	RemoteIDField                   string
	RemoteIDProvenanceField         string
	MappingFingerprint              string
	BodyFingerprint                 string
}

// preparedScopedRemoteLink is the process-local result of the exact scoped
// link read used to choose CREATE versus UPDATE. Dispatch must observe the
// same presence and, for UPDATE, the same immutable identity before provider
// I/O is allowed.
type preparedScopedRemoteLink struct {
	scope                    Scope
	commandID, payloadDigest string
	found                    bool
	link                     remoteLinkIdentity
	associationLinks         []preparedRemoteLinkSnapshot
}

type preparedRemoteLinkSnapshot struct {
	linkKey string
	found   bool
	link    remoteLinkIdentity
}

func newPreparedScopedRemoteLink(
	command Command,
	link remoteLinkIdentity,
	found bool,
	associationLinks []preparedRemoteLinkSnapshot,
) *preparedScopedRemoteLink {
	proof := &preparedScopedRemoteLink{
		scope: command.Scope, commandID: command.CommandID,
		payloadDigest: command.Intent.PayloadDigest, found: found,
		associationLinks: append([]preparedRemoteLinkSnapshot(nil), associationLinks...),
	}
	if found {
		proof.link = link
	}
	return proof
}

func lockPreparedScopedRemoteLink(ctx context.Context, tx *sql.Tx, command Command, proof *preparedScopedRemoteLink) error {
	if proof == nil || proof.scope != command.Scope || proof.commandID != command.CommandID ||
		proof.payloadDigest != command.Intent.PayloadDigest || command.Intent.Operation != outboundintentv1.OperationUpsert ||
		command.LinkKey != outboundLinkKey(command.Scope, command.Intent) {
		return ErrIdentityConflict
	}
	type snapshot struct {
		linkKey string
		found   bool
		link    remoteLinkIdentity
		primary bool
	}
	snapshots := make([]snapshot, 0, 1+len(proof.associationLinks))
	snapshots = append(snapshots, snapshot{
		linkKey: command.LinkKey, found: proof.found, link: proof.link, primary: true,
	})
	for _, prepared := range proof.associationLinks {
		snapshots = append(snapshots, snapshot{
			linkKey: prepared.linkKey, found: prepared.found, link: prepared.link,
		})
	}
	sort.SliceStable(snapshots, func(i, j int) bool { return snapshots[i].linkKey < snapshots[j].linkKey })
	for index, prepared := range snapshots {
		if prepared.linkKey == "" {
			return ErrIdentityConflict
		}
		if index > 0 && snapshots[index-1].linkKey == prepared.linkKey {
			previous := snapshots[index-1]
			if previous.found != prepared.found || (prepared.found && previous.link != prepared.link) {
				return ErrIdentityConflict
			}
			continue
		}
		link, found, err := lockRemoteLink(ctx, tx, prepared.linkKey)
		if err != nil {
			return err
		}
		if found != prepared.found || (found && link != prepared.link) {
			field := "PreparedAssociationRemoteLink"
			if prepared.primary {
				field = "PreparedRemoteLink"
			}
			return newDurableIdentityConflict(command, "dispatch_admission", field)
		}
		if !found {
			continue
		}
		valid := validRemoteLinkAuthority(command.Scope, link)
		if prepared.primary {
			valid = validDispatchRemoteLink(link, command)
		}
		if !valid {
			return newDurableIdentityConflict(command, "dispatch_admission", "RemoteLinkIdentity")
		}
	}
	return nil
}

func validateRemoteLinkFinalization(claim Claim, value RemoteLinkFinalization) error {
	if err := validateClaim(claim); err != nil || claim.Command.LinkKey == "" ||
		value.Scope != claim.Command.Scope || value.Scope.validate() != nil ||
		value.LinkKey != claim.Command.LinkKey || value.RemoteID == "" ||
		value.RemoteID != strings.TrimSpace(value.RemoteID) {
		return ErrInvalidInput
	}
	for _, field := range []string{
		value.ConnectorOwnerKey, value.ActionSlotID, value.SourceObject,
		value.SourceRecordID, value.DestinationSlotID, value.RequesterOwnerPack,
		value.ProfileOwnerPack, value.FlowDeveloperName, value.FlowActionName,
		value.TargetObject, value.ProfileName, value.PackInstallationID,
		value.CredentialRef, value.EndpointClass, value.RetryPolicyJSON,
		value.ProfileContractHash, value.CreatePathTemplate, value.UpdatePathTemplate,
		value.ResponseIDPath, value.RemoteIDField, value.MappingFingerprint,
		value.BodyFingerprint,
	} {
		if strings.TrimSpace(field) == "" || field != strings.TrimSpace(field) {
			return ErrInvalidInput
		}
	}
	intent := claim.Command.Intent
	if value.ConnectorOwnerKey != intent.ConnectorOwnerKey ||
		value.ActionSlotID != intent.ActionSlotID ||
		value.SourceObject != intent.SourceObject ||
		value.SourceRecordID != intent.SourceRecordID ||
		value.DestinationSlotID != intent.DestinationSlotID ||
		value.FlowDeveloperName != intent.FlowDeveloperName ||
		value.FlowActionName != intent.FlowActionName {
		return ErrIdentityConflict
	}
	return nil
}

func (r *Repository) succeedWithRemoteLink(
	ctx context.Context,
	claim Claim,
	status int,
	finalization RemoteLinkFinalization,
	disposition *providerSuccessDisposition,
	legacyProof *preparedLegacyRemoteLink,
) (Command, error) {
	if r == nil || r.db == nil || ctx == nil {
		return Command{}, ErrDependency
	}
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return Command{}, err
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (Command, error) {
		return Command{}, errors.Join(cause, tx.Rollback())
	}
	locked, retryCount, err := scanStoredCommand(tx.QueryRowContext(
		ctx,
		commandSelect+` WHERE Id = ? LIMIT 1 FOR UPDATE`,
		claim.Command.CommandID,
	))
	if err != nil {
		return rollback(err)
	}
	if locked.State != StateDispatching || locked.ClaimToken != claim.Token ||
		locked.LinkKey != finalization.LinkKey || locked.LeaseExpiresAt == nil ||
		!locked.LeaseExpiresAt.After(r.now().UTC()) ||
		!sameCommand(locked, claim.Command.Scope, claim.Command.Intent) {
		return rollback(ErrLeaseLost)
	}
	if legacyProof != nil {
		if finalization.Create || finalization.RemoteID != legacyProof.link.RemoteID {
			return rollback(ErrIdentityConflict)
		}
		if err := lockPreparedLegacyRemoteLink(ctx, tx, locked, legacyProof); err != nil {
			return rollback(err)
		}
	}
	link, found, err := lockRemoteLink(ctx, tx, finalization.LinkKey)
	if err != nil {
		return rollback(err)
	}
	if found {
		differences := remoteLinkIdentityDifferences(link, finalization)
		if link.Status != remoteLinkStatusFinalized {
			differences = append(differences, "Status")
		}
		if link.RemoteID != finalization.RemoteID {
			differences = append(differences, "RemoteID")
		}
		if len(differences) != 0 {
			return rollback(newRemoteLinkIdentityConflict(
				locked, link, finalization, differences...,
			))
		}
		if _, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
   SET LastValidatedPackInstallId = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
 WHERE Id = ? AND LinkKey = ? AND Status = ? AND RemoteId = ?`,
			finalization.PackInstallationID, link.ID, link.LinkKey,
			remoteLinkStatusFinalized, link.RemoteID); err != nil {
			return rollback(err)
		}
	} else {
		if !finalization.Create && legacyProof == nil {
			return rollback(newMissingRemoteLinkIdentityConflict(
				locked, finalization, "RemoteLinkPresence",
			))
		}
		if err := insertRemoteLink(ctx, tx, finalization); err != nil {
			return rollback(err)
		}
	}
	now := r.now().UTC()
	if err := wakeRemoteLinkDependents(ctx, tx, finalization, now); err != nil {
		return rollback(err)
	}
	nextState := StateWriteback
	var nextAttempt any
	var failureCode any
	if disposition != nil {
		if disposition.state != StateRetry && disposition.state != StateUncertain &&
			(disposition.state != StateDispatching || !disposition.keepClaim) {
			return rollback(ErrInvalidInput)
		}
		if disposition.minimum != nil && disposition.state != StateRetry {
			return rollback(ErrInvalidInput)
		}
		if disposition.keepClaim && (disposition.minimum != nil || disposition.code == "") {
			return rollback(ErrInvalidInput)
		}
		nextState = disposition.state
		if disposition.minimum != nil {
			seriesCount := retryCount
			if locked.RemoteID == "" {
				// The main provider effect just made durable forward progress;
				// association continuation starts its own safe-retry series.
				seriesCount = 0
			}
			nextAttempt = now.Add(safeRetryDelay(locked.CommandID, seriesCount, *disposition.minimum))
		}
		failureCode = disposition.code
	}
	if disposition != nil && disposition.keepClaim {
		result, err := tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
	SET AttemptCount = 1, RetryCount = 0,
	    FailureCode = ?, ResponseStatus = ?, RemoteId = ?, LastModifiedDate = ?
	WHERE Id = ? AND State = ? AND BINARY ClaimToken = BINARY ?
	  AND LeaseExpiresAt > ? AND DispatchStartedAt IS NOT NULL`,
			disposition.code, status, finalization.RemoteID, now,
			claim.Command.CommandID, StateDispatching, claim.Token, now)
		if err != nil {
			return rollback(err)
		}
		if affected, affectedErr := result.RowsAffected(); affectedErr != nil || affected != 1 {
			return rollback(ErrLeaseLost)
		}
		if err := tx.Commit(); err != nil {
			return Command{}, err
		}
		return r.loadByID(ctx, claim.Command.CommandID)
	}
	result, err := tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL,
	   LeaseExpiresAt = NULL, DispatchStartedAt = NULL,
	   NextAttemptAt = ?, FailureCode = ?,
	   RetryCount = CASE
	       WHEN ? <> ? THEN 0
	       WHEN RemoteId IS NULL THEN 1
	       ELSE RetryCount + 1
	   END,
       ResponseStatus = ?, RemoteId = ?, LastModifiedDate = ?
	WHERE Id = ? AND State = ? AND BINARY ClaimToken = BINARY ?
	  AND LeaseExpiresAt > ? AND DispatchStartedAt IS NOT NULL`,
		nextState, nextAttempt, failureCode, nextState, StateRetry,
		status, finalization.RemoteID, now,
		claim.Command.CommandID, StateDispatching, claim.Token, now)
	if err != nil {
		return rollback(err)
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return rollback(ErrLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return Command{}, err
	}
	return r.loadByID(ctx, claim.Command.CommandID)
}

// FinalizeWriteback preserves the released store interface and its strict
// current-link validation. Production workers use the additive receipt-aware
// finalizer after validating Runtime's exact acknowledgement.
func (r *Repository) FinalizeWriteback(ctx context.Context, claim Claim) (Command, error) {
	return r.finalizeWriteback(ctx, claim, nil)
}

func (r *Repository) finalizeWritebackWithCompletion(ctx context.Context, claim Claim, completion writebackCompletion) (Command, error) {
	return r.finalizeWriteback(ctx, claim, &completion)
}

func (r *Repository) finalizeWriteback(ctx context.Context, claim Claim, completion *writebackCompletion) (Command, error) {
	if r == nil || r.db == nil || ctx == nil || validateClaim(claim) != nil {
		return Command{}, ErrInvalidInput
	}
	typedSchema := false
	if claim.Command.RemoteID == "" {
		var err error
		typedSchema, err = r.typedResultsAvailable(ctx)
		if err != nil {
			return Command{}, err
		}
	}
	now := r.now().UTC()
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return Command{}, err
	}
	defer func() { _ = cleanup() }()
	rollback := func(cause error) (Command, error) {
		return Command{}, errors.Join(cause, tx.Rollback())
	}
	locked, err := scanCommand(tx.QueryRowContext(ctx, commandSelect+` WHERE Id = ? LIMIT 1 FOR UPDATE`, claim.Command.CommandID))
	if err != nil {
		return rollback(err)
	}
	typedResult := false
	var terminal storedTypedTerminalResult
	if typedSchema {
		var typedErr error
		terminal, typedErr = scanTypedTerminalResult(tx.QueryRowContext(ctx, `
SELECT COALESCE(ResultFormatVersion, 0), COALESCE(ResultDigest, ''), COALESCE(CanonicalResult, '')
  FROM _System_IntegrationOutboundCommand WHERE Id = ? LIMIT 1`, locked.CommandID), locked)
		switch {
		case typedErr == nil:
			typedResult = true
		case errors.Is(typedErr, ErrCommandNotFound):
		default:
			return rollback(typedErr)
		}
	}
	if locked.State != StateWriteback || locked.ClaimToken != claim.Token ||
		locked.LeaseExpiresAt == nil || !locked.LeaseExpiresAt.After(now) ||
		(typedResult && locked.RemoteID != "") || (!typedResult && locked.RemoteID == "") ||
		!sameCommand(locked, claim.Command.Scope, claim.Command.Intent) {
		return rollback(ErrLeaseLost)
	}
	if completion != nil && validateWritebackCompletion(locked, terminal, completion) != nil {
		return rollback(newDurableIdentityConflict(locked, "writeback_finalize", "RuntimeCompletionReceipt"))
	}
	if terminal.formatVersion == outboundintentv1.ResourceNoEffectResultFormatV1 && completion == nil {
		return rollback(newDurableIdentityConflict(locked, "writeback_finalize", "RuntimeCompletionReceipt"))
	}
	if !typedResult {
		link, found, err := lockRemoteLink(ctx, tx, locked.LinkKey)
		if err != nil {
			return rollback(err)
		}
		currentIdentity := found && sameRemoteLinkForCommand(link, locked) &&
			link.Status == remoteLinkStatusFinalized && link.RemoteID != "" &&
			link.RemoteID == strings.TrimSpace(link.RemoteID) && link.RemoteID == locked.RemoteID
		if !currentIdentity && completion == nil {
			return rollback(newDurableIdentityConflict(
				locked, "writeback_finalize", "CompatibilityMirrorIdentity",
			))
		}
		// Runtime's acknowledgement precedes this Integration bookkeeping.
		// Its owner fence may already have allowed a transfer or deletion. The
		// exact receipt completes this command; it never revives or rewrites a
		// retired, quarantined, missing, or changed compatibility snapshot.
		if currentIdentity {
			result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
   SET CompatibilityMirrorStatus = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
 WHERE LinkKey = ? AND Status = ? AND RemoteId = ?`,
				"synced", locked.LinkKey, remoteLinkStatusFinalized, locked.RemoteID)
			if err != nil {
				return rollback(err)
			}
			affected, err := result.RowsAffected()
			if err != nil {
				return rollback(err)
			}
			// A matched, already-synced DATETIME(0) row may change nothing within
			// the same second. Identity is proven above under the retained
			// Command -> Link locks, not by MySQL's changed-row count.
			if affected < 0 || affected > 1 {
				return rollback(newDurableIdentityConflict(
					locked, "writeback_finalize", "CompatibilityMirrorIdentity",
				))
			}
		}
	}
	terminalState, terminalCode := typedResultTerminalState(terminal)
	result, err := tx.ExecContext(ctx, `
UPDATE _System_IntegrationOutboundCommand
   SET State = ?, ClaimedBy = NULL, ClaimToken = NULL, ClaimedAt = NULL,
       LeaseExpiresAt = NULL, NextAttemptAt = NULL, FailureCode = NULLIF(?, ''), RetryCount = 0,
       LastModifiedDate = ?
 WHERE Id = ? AND State = ? AND BINARY ClaimToken = BINARY ?
   AND LeaseExpiresAt > ?`,
		terminalState, terminalCode, now, locked.CommandID, StateWriteback, claim.Token, now)
	if err != nil {
		return rollback(err)
	}
	if affected, affectedErr := result.RowsAffected(); affectedErr != nil || affected != 1 {
		return rollback(ErrLeaseLost)
	}
	if err := tx.Commit(); err != nil {
		return Command{}, err
	}
	return r.loadByID(ctx, locked.CommandID)
}

func lockRemoteLink(ctx context.Context, tx *sql.Tx, linkKey string) (remoteLinkIdentity, bool, error) {
	var value remoteLinkIdentity
	err := tx.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 FOR UPDATE`, linkKey).Scan(
		&value.ID, &value.LinkKey, &value.Status, &value.RemoteID,
		&value.ConnectorOwnerKey, &value.ActionSlotID,
		&value.RemoteLocatorHash, &value.RemoteLocatorActiveKey, &value.ConnectorNamespace,
		&value.ActionType,
		&value.SourceObject, &value.SourceRecordID, &value.DestinationSlotID,
		&value.RequesterOwnerPack, &value.ProfileOwnerPack,
		&value.FlowDeveloperName, &value.FlowActionName,
		&value.TargetObject, &value.ProfileName, &value.CreatedPackInstallationID,
		&value.LastValidatedPackInstallationID, &value.CredentialRef,
		&value.EndpointClass, &value.RetryPolicyJSON, &value.ProfileContractHash,
		&value.CreatePathTemplate, &value.UpdatePathTemplate, &value.ResponseIDPath,
		&value.RemoteIDField, &value.RemoteIDProvenanceField,
		&value.MappingFingerprint, &value.BodyFingerprint,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return remoteLinkIdentity{}, false, nil
	}
	return value, err == nil, err
}

func newDurableIdentityConflict(command Command, stage string, differingFields ...string) *DurableIdentityConflictError {
	return &DurableIdentityConflictError{
		Stage: stage, CommandID: command.CommandID, IntentID: command.Intent.IntentID,
		LinkKey: command.LinkKey, CommandState: command.State, Operation: string(command.Intent.Operation),
		SourceObject: command.Intent.SourceObject, SourceRecordID: command.Intent.SourceRecordID,
		PayloadFingerprint: command.Intent.PayloadDigest,
		DifferingFields:    append([]string(nil), differingFields...),
	}
}

func newRemoteLinkIdentityConflict(
	command Command,
	durable remoteLinkIdentity,
	candidate RemoteLinkFinalization,
	differingFields ...string,
) error {
	detail := newDurableIdentityConflict(command, "remote_link_finalize", differingFields...)
	detail.DurableIdentityFingerprint = remoteLinkIdentityFingerprint(durable)
	detail.CandidateIdentityFingerprint = remoteLinkFinalizationFingerprint(candidate)
	detail.DurableMappingFingerprint = safeConflictFingerprint("mapping", durable.MappingFingerprint)
	detail.CandidateMappingFingerprint = safeConflictFingerprint("mapping", candidate.MappingFingerprint)
	detail.DurableBodyFingerprint = safeConflictFingerprint("body", durable.BodyFingerprint)
	detail.CandidateBodyFingerprint = safeConflictFingerprint("body", candidate.BodyFingerprint)
	return detail
}

func newMissingRemoteLinkIdentityConflict(
	command Command,
	candidate RemoteLinkFinalization,
	differingFields ...string,
) error {
	detail := newDurableIdentityConflict(command, "remote_link_finalize", differingFields...)
	detail.CandidateIdentityFingerprint = remoteLinkFinalizationFingerprint(candidate)
	detail.CandidateMappingFingerprint = safeConflictFingerprint("mapping", candidate.MappingFingerprint)
	detail.CandidateBodyFingerprint = safeConflictFingerprint("body", candidate.BodyFingerprint)
	return detail
}

func remoteLinkIdentityFingerprint(value remoteLinkIdentity) string {
	return safeConflictFingerprint("remote-link-identity", remoteLinkComparableFields(value)...)
}

func remoteLinkFinalizationFingerprint(value RemoteLinkFinalization) string {
	return safeConflictFingerprint("remote-link-identity", remoteLinkFinalizationComparableFields(value)...)
}

func remoteLinkComparableFields(value remoteLinkIdentity) []string {
	return []string{
		value.LinkKey, value.ConnectorOwnerKey, value.ActionSlotID, value.ActionType,
		value.SourceObject, value.SourceRecordID, value.DestinationSlotID,
		value.RequesterOwnerPack, value.ProfileOwnerPack, value.ConnectorNamespace,
		value.RemoteLocatorHash, value.RemoteLocatorActiveKey, value.FlowDeveloperName,
		value.FlowActionName, value.TargetObject, value.ProfileName,
		value.LastValidatedPackInstallationID, value.CredentialRef, value.EndpointClass,
		value.RetryPolicyJSON, value.ProfileContractHash, value.CreatePathTemplate,
		value.UpdatePathTemplate, value.ResponseIDPath, value.RemoteIDField,
		value.RemoteIDProvenanceField, value.MappingFingerprint, value.BodyFingerprint,
		value.Status, value.RemoteID,
	}
}

func remoteLinkFinalizationComparableFields(value RemoteLinkFinalization) []string {
	locator := remoteLocatorHash(value.Scope, value.ConnectorOwnerKey, value.DestinationSlotID, value.RemoteID)
	return []string{
		value.LinkKey, value.ConnectorOwnerKey, value.ActionSlotID, connectorPushActionType,
		value.SourceObject, value.SourceRecordID, value.DestinationSlotID,
		value.RequesterOwnerPack, value.ProfileOwnerPack, value.ConnectorOwnerKey,
		locator, locator, value.FlowDeveloperName,
		value.FlowActionName, value.TargetObject, value.ProfileName,
		value.PackInstallationID, value.CredentialRef, value.EndpointClass,
		value.RetryPolicyJSON, value.ProfileContractHash, value.CreatePathTemplate,
		value.UpdatePathTemplate, value.ResponseIDPath, value.RemoteIDField,
		value.RemoteIDProvenanceField, value.MappingFingerprint, value.BodyFingerprint,
		remoteLinkStatusFinalized, value.RemoteID,
	}
}

func safeConflictFingerprint(domain string, values ...string) string {
	sum := sha256.Sum256([]byte(domain + "\x00" + strings.Join(values, "\x1f")))
	return "sha256:" + hex.EncodeToString(sum[:])
}

func remoteLinkIdentityDifferences(link remoteLinkIdentity, value RemoteLinkFinalization) []string {
	differences := make([]string, 0, 8)
	locator := remoteLocatorHash(value.Scope, value.ConnectorOwnerKey, value.DestinationSlotID, value.RemoteID)
	for _, candidate := range []struct {
		name  string
		equal bool
	}{
		{name: "LinkKey", equal: link.LinkKey == value.LinkKey},
		{name: "ConnectorOwnerKey", equal: link.ConnectorOwnerKey == value.ConnectorOwnerKey},
		{name: "ActionSlotID", equal: link.ActionSlotID == value.ActionSlotID},
		{name: "ActionType", equal: link.ActionType == connectorPushActionType},
		{name: "SourceObject", equal: link.SourceObject == value.SourceObject},
		{name: "SourceRecordID", equal: link.SourceRecordID == value.SourceRecordID},
		{name: "DestinationSlotID", equal: link.DestinationSlotID == value.DestinationSlotID},
		{name: "RequesterOwnerPack", equal: link.RequesterOwnerPack == value.RequesterOwnerPack},
		{name: "ProfileOwnerPack", equal: link.ProfileOwnerPack == value.ProfileOwnerPack},
		{name: "ConnectorNamespace", equal: link.ConnectorNamespace == value.ConnectorOwnerKey},
		{name: "RemoteLocatorHash", equal: link.RemoteLocatorHash == locator},
		{name: "RemoteLocatorActiveKey", equal: link.RemoteLocatorActiveKey == locator},
		{name: "FlowDeveloperName", equal: link.FlowDeveloperName == value.FlowDeveloperName},
		{name: "FlowActionName", equal: link.FlowActionName == value.FlowActionName},
		{name: "TargetObject", equal: link.TargetObject == value.TargetObject},
		{name: "ProfileName", equal: link.ProfileName == value.ProfileName},
		{name: "LastValidatedPackInstallationID", equal: link.LastValidatedPackInstallationID == value.PackInstallationID},
		{name: "CredentialRef", equal: link.CredentialRef == value.CredentialRef},
		{name: "EndpointClass", equal: link.EndpointClass == value.EndpointClass},
		{name: "RetryPolicyJSON", equal: link.RetryPolicyJSON == value.RetryPolicyJSON},
		{name: "ProfileContractHash", equal: link.ProfileContractHash == value.ProfileContractHash},
		{name: "CreatePathTemplate", equal: link.CreatePathTemplate == value.CreatePathTemplate},
		{name: "UpdatePathTemplate", equal: link.UpdatePathTemplate == value.UpdatePathTemplate},
		{name: "ResponseIDPath", equal: link.ResponseIDPath == value.ResponseIDPath},
		{name: "RemoteIDField", equal: link.RemoteIDField == value.RemoteIDField},
		{name: "RemoteIDProvenanceField", equal: link.RemoteIDProvenanceField == value.RemoteIDProvenanceField},
		{name: "MappingFingerprint", equal: link.MappingFingerprint == value.MappingFingerprint},
		{name: "BodyFingerprint", equal: link.BodyFingerprint == value.BodyFingerprint},
	} {
		if !candidate.equal {
			differences = append(differences, candidate.name)
		}
	}
	return differences
}

func insertRemoteLink(ctx context.Context, tx *sql.Tx, value RemoteLinkFinalization) error {
	locator := remoteLocatorHash(value.Scope, value.ConnectorOwnerKey, value.DestinationSlotID, value.RemoteID)
	_, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorRemoteLink
  (Id, LinkKey, Status, RemoteId, RemoteLocatorHash, RemoteLocatorActiveKey,
   ConnectorOwnerKey, ActionSlotId, ActionType, SourceObject, SourceRecordId,
   DestinationSlotId, RequesterOwnerPack, ProfileOwnerPack, ConnectorNamespace,
   FlowDeveloperName, FlowActionName, TargetObject, ProfileName,
   CreatedPackInstallId, LastValidatedPackInstallId, CredentialRef, EndpointClass,
   RetryPolicy, ProfileContractHash, CreatePathTemplate, UpdatePathTemplate,
   ResponseIdPath, RemoteIdField, RemoteIdProvenanceField, MappingFingerprint,
   BodyFingerprint, CompatibilityMirrorStatus)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
        ?, ?, ?, ?, ?, ?, NULLIF(?, ''), ?, ?, ?)`,
		remoteLinkID(value.LinkKey), value.LinkKey, remoteLinkStatusFinalized,
		value.RemoteID, locator, locator, value.ConnectorOwnerKey, value.ActionSlotID,
		connectorPushActionType, value.SourceObject, value.SourceRecordID,
		value.DestinationSlotID, value.RequesterOwnerPack, value.ProfileOwnerPack,
		// ConnectorNamespace is the established durable snapshot lookup key,
		// despite its legacy name. The neutral outbound contract intentionally
		// carries no authenticated inbound route namespace; Core writes the same
		// ConnectorOwnerKey value when composing this snapshot.
		value.ConnectorOwnerKey, value.FlowDeveloperName, value.FlowActionName,
		value.TargetObject, value.ProfileName, value.PackInstallationID,
		value.PackInstallationID, value.CredentialRef, value.EndpointClass,
		value.RetryPolicyJSON, value.ProfileContractHash, value.CreatePathTemplate,
		value.UpdatePathTemplate, value.ResponseIDPath, value.RemoteIDField,
		value.RemoteIDProvenanceField, value.MappingFingerprint,
		value.BodyFingerprint, remoteLinkMirrorPending)
	if err != nil {
		return fmt.Errorf("connector outbound intent finalize remote link: %w", err)
	}
	return nil
}

func remoteLinkID(linkKey string) string {
	sum := sha256.Sum256([]byte("integration-remote-link\x00" + linkKey))
	return "integration-link-" + hex.EncodeToString(sum[:])
}

func remoteLocatorHash(scope Scope, connectorOwnerKey, destinationSlotID, remoteID string) string {
	return outboundintentv1.IntegrationRemoteLocatorHash(integrationLogicalEffectScope(scope), connectorOwnerKey, destinationSlotID, remoteID)
}