package automation

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"

	"github.com/nexuscrm/kernel/pkg/ports"
)

const ConnectorInboundRemoteIdentityRecoveryBatchMaxItems = 100

const (
	ConnectorInboundRemoteIdentityRecoveryBatchOutcomeDryRunVerified           = "dry_run_verified"
	ConnectorInboundRemoteIdentityRecoveryBatchOutcomeCommittedVerified        = "committed_verified"
	ConnectorInboundRemoteIdentityRecoveryBatchOutcomeCommittedReadbackBlocked = "committed_readback_blocked"
	ConnectorInboundRemoteIdentityRecoveryBatchOutcomeRolledBackBlocked        = "rolled_back_blocked"
)

var (
	errConnectorInboundRemoteIdentityRecoveryDryRunRollback       = errors.New("connector inbound remote identity recovery dry-run rollback")
	ErrConnectorInboundRemoteIdentityRecoveryBatchReadbackBlocked = errors.New("connector inbound remote identity recovery batch committed but authoritative readback blocked")
)

// ConnectorInboundRemoteIdentityRecoverySelector contains only authority
// selectors derived by a trusted caller. DestinationSlotID scopes RemoteID to
// the installed destination; it is not inferred from inbound evidence.
type ConnectorInboundRemoteIdentityRecoverySelector struct {
	ConnectorOwnerKey  string
	ConnectorNamespace string
	DestinationSlotID  string
	RemoteID           string
}

// ConnectorInboundRemoteIDEvidenceBinding is one installed source binding in
// which inverse evidence discovery is permitted. A remote ID is never searched
// globally because different source-object ID spaces may overlap.
type ConnectorInboundRemoteIDEvidenceBinding struct {
	SourceObject  string
	RemoteIDField string
}

// ConnectorInboundRemoteIdentityRecoveryBinding pairs an allowed evidence
// binding with the current installed durable identity and snapshot. The source
// record ID is deliberately absent from the installed identity and is filled
// only from the exact connector-authored evidence row selected in the recovery
// transaction.
type ConnectorInboundRemoteIdentityRecoveryBinding struct {
	EvidenceBinding           ConnectorInboundRemoteIDEvidenceBinding
	Identity                  ConnectorPushDurableIdentity
	Snapshot                  ConnectorPushDurableSnapshot
	RecoverySourceEligibility []string
}

type ConnectorInboundRemoteIdentityRecoveryRequest struct {
	Selector ConnectorInboundRemoteIdentityRecoverySelector
}

// ConnectorInboundRemoteIdentityRecoveryBatchSelector is shared by every item
// in one bounded maintenance transaction. The remote ID stays item-owned so the
// engine can derive and sort the exact durable locator lanes before mutation.
type ConnectorInboundRemoteIdentityRecoveryBatchSelector struct {
	ConnectorOwnerKey  string
	ConnectorNamespace string
	DestinationSlotID  string
}

// ConnectorInboundRemoteIdentityRecoveryBatchItem carries no durable
// authority. Recovery still selects the actual source exclusively from locked
// installed metadata and connector-authored evidence. The expected source pair
// is checked only after that strict ambient transition has selected and
// finalized the actual pair, so a mismatch rolls the whole batch back.
type ConnectorInboundRemoteIdentityRecoveryBatchItem struct {
	RemoteID               string
	ExpectedSourceObject   string
	ExpectedSourceRecordID string
}

type ConnectorInboundRemoteIdentityRecoveryBatchRequest struct {
	Selector ConnectorInboundRemoteIdentityRecoveryBatchSelector
	Items    []ConnectorInboundRemoteIdentityRecoveryBatchItem
}

// ConnectorInboundRemoteIdentityRecoveryInstalledBindings is resolved from one
// locked, active, converged installed-pack generation. Generation is retained
// as an explicit fail-closed proof that the resolver did not assemble bindings
// from an unlocked or drifting metadata view.
type ConnectorInboundRemoteIdentityRecoveryInstalledBindings struct {
	Generation string
	Bindings   []ConnectorInboundRemoteIdentityRecoveryBinding
}

// ConnectorInboundRemoteIdentityRecoveryInstalledResolver is a required
// ambient-transaction port. The implementation must lock the installed pack
// generation before returning current contract/profile/mapping-derived
// bindings. Operator input is never accepted as durable authority.
type ConnectorInboundRemoteIdentityRecoveryInstalledResolver interface {
	ResolveConnectorInboundRemoteIdentityRecoveryInstalledBindings(
		ctx context.Context,
		selector ConnectorInboundRemoteIdentityRecoverySelector,
	) (ConnectorInboundRemoteIdentityRecoveryInstalledBindings, error)
}

// ConnectorInboundRemoteIdentityRecoveryResult is returned only after the
// self-owned transaction commits. A nil result with nil error means that no
// unique authored or consumed retry evidence exists inside the installed
// binding allowlist.
type ConnectorInboundRemoteIdentityRecoveryResult struct {
	EvidenceID     string
	LinkKey        string
	SourceObject   string
	SourceRecordID string
	RemoteIDField  string
}

type ConnectorInboundRemoteIdentityRecoveryBatchItemResult struct {
	RemoteID       string
	EvidenceID     string
	LinkKey        string
	SourceObject   string
	SourceRecordID string
	RemoteIDField  string
}

// ConnectorInboundRemoteIdentityRecoveryBatchResult distinguishes a
// transaction that never committed from a commit whose subsequent
// authoritative resolver readback blocked downstream enrollment. A readback
// failure can never truthfully be reported as a database rollback.
type ConnectorInboundRemoteIdentityRecoveryBatchResult struct {
	Outcome           string
	Committed         bool
	ReadbackVerified  bool
	EnrollmentAllowed bool
	RequestedCount    int
	RecoveredCount    int
	Items             []ConnectorInboundRemoteIdentityRecoveryBatchItemResult
	BlockReason       string
}

// ConnectorInboundRemoteIdentityRecoverySourceValidation is produced by a
// locked live-record read. Only synced and drifted are publishable; pending or
// arbitrary caller strings fail closed.
type ConnectorInboundRemoteIdentityRecoverySourceValidation struct {
	CompatibilityMirrorStatus ConnectorPushCompatibilityMirrorStatus
}

// ConnectorInboundRemoteIdentityRecoverySourceValidator is intentionally a
// required port. PR-A owns transaction and durable-state correctness; the
// installed binding implementation owns generic live-record lifecycle and
// mirror readback without hard-coding an application object such as Lead.
type ConnectorInboundRemoteIdentityRecoverySourceValidator interface {
	ValidateConnectorInboundRemoteIdentityRecoverySource(
		ctx context.Context,
		evidence ConnectorInboundRemoteIDEvidence,
		binding ConnectorInboundRemoteIdentityRecoveryBinding,
	) (ConnectorInboundRemoteIdentityRecoverySourceValidation, error)
}

type connectorInboundRemoteIdentityRecoveryTransactionManager interface {
	RunInPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error
	IsTransactionActive(ctx context.Context) bool
}

type connectorInboundRemoteIdentityRecoveryOutcomeTransactionManager interface {
	RunInPessimisticTransactionWithOutcome(
		ctx context.Context,
		fn func(context.Context) error,
	) (ports.TransactionCompletionOutcome, error)
}

var _ connectorInboundRemoteIdentityRecoveryTransactionManager = (PersistenceProvider)(nil)

type connectorInboundRemoteIdentityRecoveryStore interface {
	FindInboundRemoteIDEvidenceHintsInAmbientTx(
		ctx context.Context,
		selector ConnectorInboundRemoteIdentityRecoverySelector,
	) ([]ConnectorInboundRemoteIDEvidence, error)
	FindUniqueInboundRemoteIDEvidenceForRecoveryInAmbientTx(
		ctx context.Context,
		selector ConnectorInboundRemoteIdentityRecoverySelector,
		bindings []ConnectorInboundRemoteIDEvidenceBinding,
	) (*ConnectorInboundRemoteIDEvidence, error)
	LockConnectorRemoteIdentityLaneInAmbientTx(
		ctx context.Context,
		identity ConnectorPushDurableIdentity,
		remoteID string,
	) error
	FinalizeInboundRemoteIdentityInAmbientTx(ctx context.Context, finalization ConnectorInboundRemoteIdentityFinalization) error
	AssertExactInboundRemoteIDEvidenceRecoveryInAmbientTx(
		ctx context.Context,
		selector ConnectorInboundRemoteIdentityRecoverySelector,
		bindings []ConnectorInboundRemoteIDEvidenceBinding,
		evidenceID string,
		consumedByLinkKey string,
	) error
	CheckConnectorRemoteIdentityRecoveryHistoryInAmbientTx(
		ctx context.Context,
		finalization ConnectorInboundRemoteIdentityFinalization,
	) (alreadyFinalized bool, err error)
	AssertExactConnectorRemoteIdentityRecoveryHistoryInAmbientTx(
		ctx context.Context,
		finalization ConnectorInboundRemoteIdentityFinalization,
	) error
	ReadConnectorInboundRemoteIdentityRecoveryEvidence(
		ctx context.Context,
		evidenceID string,
	) (*ConnectorInboundRemoteIDEvidence, error)
}

// ConnectorInboundRemoteIdentityRecoveryService owns the short pessimistic
// recovery transaction. Its installed-generation-before-record lock order can
// oppose a conversion Flow that already holds record locks, so both sides own
// complete pessimistic retry callbacks: TiDB 1213 deadlock or 1205 lock-wait
// contention replays the whole recovery or conversion attempt, never only the
// inner transfer. It leaves the ordinary resolver pure and reuses the
// exact-evidence finalization transition rather than the weaker adoption API.
type ConnectorInboundRemoteIdentityRecoveryService struct {
	txManager       connectorInboundRemoteIdentityRecoveryTransactionManager
	store           connectorInboundRemoteIdentityRecoveryStore
	installed       ConnectorInboundRemoteIdentityRecoveryInstalledResolver
	sourceValidator ConnectorInboundRemoteIdentityRecoverySourceValidator
}

func NewConnectorInboundRemoteIdentityRecoveryService(
	txManager connectorInboundRemoteIdentityRecoveryTransactionManager,
	store connectorInboundRemoteIdentityRecoveryStore,
	installed ConnectorInboundRemoteIdentityRecoveryInstalledResolver,
	sourceValidator ConnectorInboundRemoteIdentityRecoverySourceValidator,
) *ConnectorInboundRemoteIdentityRecoveryService {
	return &ConnectorInboundRemoteIdentityRecoveryService{
		txManager:       txManager,
		store:           store,
		installed:       installed,
		sourceValidator: sourceValidator,
	}
}

func (s *ConnectorInboundRemoteIdentityRecoveryService) Recover(
	ctx context.Context,
	request ConnectorInboundRemoteIdentityRecoveryRequest,
) (*ConnectorInboundRemoteIdentityRecoveryResult, error) {
	if err := s.validateWiring(); err != nil {
		return nil, err
	}
	if s.txManager.IsTransactionActive(ctx) {
		return nil, fmt.Errorf("connector inbound remote identity recovery rejects an ambient transaction")
	}

	var result *ConnectorInboundRemoteIdentityRecoveryResult
	err := s.txManager.RunInPessimisticTransaction(ctx, func(txCtx context.Context) error {
		var err error
		result, err = s.recoverInAmbientTx(txCtx, request)
		return err
	})
	if err != nil {
		return nil, err
	}
	return result, nil
}

// RecoverBatch promotes every requested authored-evidence row in one
// service-owned pessimistic transaction, then verifies every committed link
// through the same finalized-locator authority used by the pure inbound
// resolver. Any transactional item failure rolls all items back. A post-commit
// readback failure instead returns a committed-but-blocked result and a typed
// error so callers cannot enroll downstream work on unverified authority.
func (s *ConnectorInboundRemoteIdentityRecoveryService) RecoverBatch(
	ctx context.Context,
	request ConnectorInboundRemoteIdentityRecoveryBatchRequest,
) (ConnectorInboundRemoteIdentityRecoveryBatchResult, error) {
	return s.recoverBatch(ctx, request, false)
}

// DryRunBatch executes the exact RecoverBatch mutation path and deliberately
// returns a private sentinel from inside the transaction to force rollback.
// It is not a separate approximate preflight.
func (s *ConnectorInboundRemoteIdentityRecoveryService) DryRunBatch(
	ctx context.Context,
	request ConnectorInboundRemoteIdentityRecoveryBatchRequest,
) (ConnectorInboundRemoteIdentityRecoveryBatchResult, error) {
	return s.recoverBatch(ctx, request, true)
}

func (s *ConnectorInboundRemoteIdentityRecoveryService) recoverBatch(
	ctx context.Context,
	request ConnectorInboundRemoteIdentityRecoveryBatchRequest,
	dryRun bool,
) (ConnectorInboundRemoteIdentityRecoveryBatchResult, error) {
	if err := s.validateWiring(); err != nil {
		return connectorInboundRemoteIdentityRecoveryBatchBlocked(0, err), err
	}
	if s.txManager.IsTransactionActive(ctx) {
		err := fmt.Errorf("connector inbound remote identity recovery batch rejects an ambient transaction")
		return connectorInboundRemoteIdentityRecoveryBatchBlocked(len(request.Items), err), err
	}
	normalized, err := normalizeConnectorInboundRemoteIdentityRecoveryBatchRequest(request)
	if err != nil {
		return connectorInboundRemoteIdentityRecoveryBatchBlocked(len(request.Items), err), err
	}

	outcomeManager, ok := s.txManager.(connectorInboundRemoteIdentityRecoveryOutcomeTransactionManager)
	if !ok || outcomeManager == nil {
		err := fmt.Errorf("connector inbound remote identity recovery typed transaction outcome manager is not wired")
		return connectorInboundRemoteIdentityRecoveryBatchBlocked(len(normalized.Items), err), err
	}

	var transactionItems []ConnectorInboundRemoteIdentityRecoveryBatchItemResult
	callbackCompleted := false
	completion, txErr := outcomeManager.RunInPessimisticTransactionWithOutcome(ctx, func(txCtx context.Context) error {
		// The transaction manager may replay this callback after a deadlock or
		// lock timeout. Keep attempt-local results so a retry cannot duplicate
		// receipt rows from an aborted attempt.
		callbackCompleted = false
		transactionItems = nil
		attemptItems := make([]ConnectorInboundRemoteIdentityRecoveryBatchItemResult, 0, len(normalized.Items))
		for _, item := range normalized.Items {
			result, recoverErr := s.recoverInAmbientTx(txCtx, ConnectorInboundRemoteIdentityRecoveryRequest{
				Selector: ConnectorInboundRemoteIdentityRecoverySelector{
					ConnectorOwnerKey:  normalized.Selector.ConnectorOwnerKey,
					ConnectorNamespace: normalized.Selector.ConnectorNamespace,
					DestinationSlotID:  normalized.Selector.DestinationSlotID,
					RemoteID:           item.RemoteID,
				},
			})
			if recoverErr != nil {
				return fmt.Errorf("recover remote ID %q: %w", item.RemoteID, recoverErr)
			}
			if result == nil {
				return fmt.Errorf("recover remote ID %q: no unique connector-authored evidence was found", item.RemoteID)
			}
			// This pair is an assertion over the actual authority selected above;
			// it never participates in evidence or installed-binding selection.
			if result.SourceObject != item.ExpectedSourceObject || result.SourceRecordID != item.ExpectedSourceRecordID {
				return fmt.Errorf(
					"recover remote ID %q: actual source %s/%s does not match expected source %s/%s",
					item.RemoteID,
					result.SourceObject,
					result.SourceRecordID,
					item.ExpectedSourceObject,
					item.ExpectedSourceRecordID,
				)
			}
			attemptItems = append(attemptItems, ConnectorInboundRemoteIdentityRecoveryBatchItemResult{
				RemoteID:       item.RemoteID,
				EvidenceID:     result.EvidenceID,
				LinkKey:        result.LinkKey,
				SourceObject:   result.SourceObject,
				SourceRecordID: result.SourceRecordID,
				RemoteIDField:  result.RemoteIDField,
			})
			transactionItems = append([]ConnectorInboundRemoteIdentityRecoveryBatchItemResult(nil), attemptItems...)
		}
		callbackCompleted = true
		if dryRun {
			return errConnectorInboundRemoteIdentityRecoveryDryRunRollback
		}
		return nil
	})
	if dryRun && completion == ports.TransactionCompletionRolledBack &&
		errors.Is(txErr, errConnectorInboundRemoteIdentityRecoveryDryRunRollback) {
		return ConnectorInboundRemoteIdentityRecoveryBatchResult{
			Outcome:        ConnectorInboundRemoteIdentityRecoveryBatchOutcomeDryRunVerified,
			RequestedCount: len(normalized.Items),
			RecoveredCount: len(transactionItems),
			Items:          transactionItems,
		}, nil
	}
	if completion == ports.TransactionCompletionRolledBack {
		if txErr == nil {
			txErr = fmt.Errorf("connector inbound remote identity recovery transaction rolled back without an error")
		}
		return connectorInboundRemoteIdentityRecoveryBatchBlocked(len(normalized.Items), txErr), txErr
	}
	if txErr != nil || completion != ports.TransactionCompletionCommitted {
		if txErr == nil {
			txErr = fmt.Errorf("connector inbound remote identity recovery transaction has unknown completion outcome %q", completion)
		}
		return s.resolveAmbiguousBatchOutcome(ctx, normalized, transactionItems, callbackCompleted, dryRun, txErr)
	}

	result := ConnectorInboundRemoteIdentityRecoveryBatchResult{
		Outcome:        ConnectorInboundRemoteIdentityRecoveryBatchOutcomeCommittedVerified,
		Committed:      true,
		RequestedCount: len(normalized.Items),
		RecoveredCount: len(transactionItems),
		Items:          transactionItems,
	}
	if err := s.verifyCommittedBatchReadback(ctx, normalized, transactionItems); err != nil {
		result.Outcome = ConnectorInboundRemoteIdentityRecoveryBatchOutcomeCommittedReadbackBlocked
		result.BlockReason = err.Error()
		return result, fmt.Errorf("%w: %v", ErrConnectorInboundRemoteIdentityRecoveryBatchReadbackBlocked, err)
	}
	result.ReadbackVerified = true
	result.EnrollmentAllowed = true
	return result, nil
}

type connectorInboundRemoteIdentityRecoveryDurableState uint8

const (
	connectorInboundRemoteIdentityRecoveryDurableUnknown connectorInboundRemoteIdentityRecoveryDurableState = iota
	connectorInboundRemoteIdentityRecoveryDurableCommitted
	connectorInboundRemoteIdentityRecoveryDurableRolledBack
)

func (s *ConnectorInboundRemoteIdentityRecoveryService) resolveAmbiguousBatchOutcome(
	ctx context.Context,
	request ConnectorInboundRemoteIdentityRecoveryBatchRequest,
	transactionItems []ConnectorInboundRemoteIdentityRecoveryBatchItemResult,
	callbackCompleted bool,
	dryRun bool,
	txErr error,
) (ConnectorInboundRemoteIdentityRecoveryBatchResult, error) {
	committedItems := make([]ConnectorInboundRemoteIdentityRecoveryBatchItemResult, 0, len(transactionItems))
	readbackUnknown := false
	for i, item := range transactionItems {
		if i >= len(request.Items) || item.RemoteID != request.Items[i].RemoteID {
			readbackUnknown = true
			continue
		}
		state, err := s.readConnectorInboundRemoteIdentityRecoveryDurableState(ctx, request, request.Items[i], item)
		if err != nil {
			readbackUnknown = true
			continue
		}
		switch state {
		case connectorInboundRemoteIdentityRecoveryDurableCommitted:
			committedItems = append(committedItems, item)
		case connectorInboundRemoteIdentityRecoveryDurableRolledBack:
			// An immediate all-zero snapshot is not positive rollback proof for
			// CommitError: an applied commit may become visible only after the
			// acknowledgement path fails. Only the transaction manager's typed
			// TransactionCompletionRolledBack outcome may authorize a rollback
			// claim; authored/no-link readback remains commit-unknown here.
		default:
			readbackUnknown = true
		}
	}

	fullCandidateSet := callbackCompleted && len(transactionItems) == len(request.Items)
	if !dryRun && fullCandidateSet && !readbackUnknown && len(committedItems) == len(request.Items) {
		return ConnectorInboundRemoteIdentityRecoveryBatchResult{
			Outcome:           ConnectorInboundRemoteIdentityRecoveryBatchOutcomeCommittedVerified,
			Committed:         true,
			ReadbackVerified:  true,
			EnrollmentAllowed: true,
			RequestedCount:    len(request.Items),
			RecoveredCount:    len(committedItems),
			Items:             committedItems,
		}, nil
	}
	result := ConnectorInboundRemoteIdentityRecoveryBatchResult{
		Outcome:        ConnectorInboundRemoteIdentityRecoveryBatchOutcomeCommittedReadbackBlocked,
		Committed:      true,
		RequestedCount: len(request.Items),
		RecoveredCount: len(committedItems),
		Items:          committedItems,
		BlockReason:    "transaction completion was not authoritative and exact durable readback could not prove one complete outcome",
	}
	return result, fmt.Errorf("%w: transaction outcome %q: %v", ErrConnectorInboundRemoteIdentityRecoveryBatchReadbackBlocked, completionOutcomeLabel(dryRun), txErr)
}

func completionOutcomeLabel(dryRun bool) string {
	if dryRun {
		return "dry_run_rollback_unknown"
	}
	return "commit_unknown"
}

func (s *ConnectorInboundRemoteIdentityRecoveryService) verifyCommittedBatchReadback(
	ctx context.Context,
	request ConnectorInboundRemoteIdentityRecoveryBatchRequest,
	items []ConnectorInboundRemoteIdentityRecoveryBatchItemResult,
) error {
	if len(items) != len(request.Items) {
		return fmt.Errorf("committed item count %d does not match requested count %d", len(items), len(request.Items))
	}
	for i, item := range items {
		expected := request.Items[i]
		if item.RemoteID != expected.RemoteID {
			return fmt.Errorf("committed result order changed at item %d", i)
		}
		state, err := s.readConnectorInboundRemoteIdentityRecoveryDurableState(ctx, request, expected, item)
		if err != nil {
			return fmt.Errorf("read remote ID %q durable recovery state: %w", item.RemoteID, err)
		}
		if state != connectorInboundRemoteIdentityRecoveryDurableCommitted {
			return fmt.Errorf("remote ID %q finalized authority and consumed evidence do not match the exact committed source", item.RemoteID)
		}
	}
	return nil
}

func (s *ConnectorInboundRemoteIdentityRecoveryService) readConnectorInboundRemoteIdentityRecoveryDurableState(
	ctx context.Context,
	request ConnectorInboundRemoteIdentityRecoveryBatchRequest,
	expected ConnectorInboundRemoteIdentityRecoveryBatchItem,
	item ConnectorInboundRemoteIdentityRecoveryBatchItemResult,
) (connectorInboundRemoteIdentityRecoveryDurableState, error) {
	resolver, ok := s.store.(ConnectorRemoteIdentityAuthorityStore)
	if !ok || resolver == nil {
		return connectorInboundRemoteIdentityRecoveryDurableUnknown, fmt.Errorf("authoritative finalized-locator resolver is not wired")
	}
	authority, err := resolver.ResolveFinalizedRemoteLocator(ctx, ConnectorRemoteLocatorQuery{
		ConnectorOwnerKey: request.Selector.ConnectorOwnerKey,
		DestinationSlotID: request.Selector.DestinationSlotID,
		RemoteID:          item.RemoteID,
	})
	if err != nil {
		return connectorInboundRemoteIdentityRecoveryDurableUnknown, err
	}
	evidence, err := s.store.ReadConnectorInboundRemoteIdentityRecoveryEvidence(ctx, item.EvidenceID)
	if err != nil {
		return connectorInboundRemoteIdentityRecoveryDurableUnknown, err
	}
	if evidence == nil ||
		strings.TrimSpace(evidence.ID) != item.EvidenceID ||
		strings.TrimSpace(evidence.ConnectorOwnerKey) != request.Selector.ConnectorOwnerKey ||
		strings.TrimSpace(evidence.ConnectorNamespace) != request.Selector.ConnectorNamespace ||
		strings.TrimSpace(evidence.SourceObject) != expected.ExpectedSourceObject ||
		strings.TrimSpace(evidence.SourceRecordId) != expected.ExpectedSourceRecordID ||
		strings.TrimSpace(evidence.RemoteIDField) != item.RemoteIDField ||
		strings.TrimSpace(evidence.RemoteId) != item.RemoteID {
		return connectorInboundRemoteIdentityRecoveryDurableUnknown, nil
	}

	if authority == nil {
		expectedActiveKey := ConnectorInboundEvidenceActiveKey(
			evidence.ConnectorOwnerKey,
			evidence.SourceObject,
			evidence.SourceRecordId,
			evidence.RemoteIDField,
		)
		if strings.TrimSpace(evidence.EvidenceStatus) == ConnectorInboundEvidenceStatusAuthored &&
			strings.TrimSpace(evidence.ConsumedByLinkKey) == "" &&
			strings.TrimSpace(evidence.ActiveKey) == expectedActiveKey {
			return connectorInboundRemoteIdentityRecoveryDurableRolledBack, nil
		}
		return connectorInboundRemoteIdentityRecoveryDurableUnknown, nil
	}

	expectedHash := ConnectorPushRemoteLocatorHash(ConnectorPushDurableIdentity{
		ConnectorOwnerKey: request.Selector.ConnectorOwnerKey,
		DestinationSlotID: request.Selector.DestinationSlotID,
	}, item.RemoteID)
	if authority.LinkStatus != ConnectorPushLinkStatusFinalized ||
		authority.LinkKey != item.LinkKey ||
		authority.RemoteID != item.RemoteID ||
		authority.RemoteLocatorHash != expectedHash ||
		authority.RemoteIDField != item.RemoteIDField ||
		authority.Identity.ConnectorOwnerKey != request.Selector.ConnectorOwnerKey ||
		authority.Identity.DestinationSlotID != request.Selector.DestinationSlotID ||
		authority.Identity.SourceObject != expected.ExpectedSourceObject ||
		authority.Identity.SourceRecordID != expected.ExpectedSourceRecordID ||
		strings.TrimSpace(evidence.EvidenceStatus) != ConnectorInboundEvidenceStatusConsumed ||
		strings.TrimSpace(evidence.ConsumedByLinkKey) != item.LinkKey ||
		strings.TrimSpace(evidence.ActiveKey) != "" {
		return connectorInboundRemoteIdentityRecoveryDurableUnknown, nil
	}
	return connectorInboundRemoteIdentityRecoveryDurableCommitted, nil
}

func connectorInboundRemoteIdentityRecoveryBatchBlocked(requested int, err error) ConnectorInboundRemoteIdentityRecoveryBatchResult {
	reason := ""
	if err != nil {
		reason = err.Error()
	}
	return ConnectorInboundRemoteIdentityRecoveryBatchResult{
		Outcome:        ConnectorInboundRemoteIdentityRecoveryBatchOutcomeRolledBackBlocked,
		RequestedCount: requested,
		BlockReason:    reason,
	}
}

func (s *ConnectorInboundRemoteIdentityRecoveryService) validateWiring() error {
	if s == nil || s.txManager == nil {
		return fmt.Errorf("connector inbound remote identity recovery transaction manager is not wired")
	}
	if s.store == nil {
		return fmt.Errorf("connector inbound remote identity recovery durable store is not wired")
	}
	if s.installed == nil {
		return fmt.Errorf("connector inbound remote identity recovery installed binding resolver is not wired")
	}
	if s.sourceValidator == nil {
		return fmt.Errorf("connector inbound remote identity recovery source validator is not wired")
	}
	return nil
}

func normalizeConnectorInboundRemoteIdentityRecoveryBatchRequest(
	request ConnectorInboundRemoteIdentityRecoveryBatchRequest,
) (ConnectorInboundRemoteIdentityRecoveryBatchRequest, error) {
	request.Selector.ConnectorOwnerKey = strings.TrimSpace(request.Selector.ConnectorOwnerKey)
	request.Selector.ConnectorNamespace = strings.TrimSpace(request.Selector.ConnectorNamespace)
	request.Selector.DestinationSlotID = strings.TrimSpace(request.Selector.DestinationSlotID)
	if request.Selector.ConnectorOwnerKey == "" || request.Selector.ConnectorNamespace == "" || request.Selector.DestinationSlotID == "" {
		return request, fmt.Errorf("connector inbound remote identity recovery batch owner, namespace, and destination slot are required")
	}
	if len(request.Items) == 0 || len(request.Items) > ConnectorInboundRemoteIdentityRecoveryBatchMaxItems {
		return request, fmt.Errorf("connector inbound remote identity recovery batch item count must be between 1 and %d", ConnectorInboundRemoteIdentityRecoveryBatchMaxItems)
	}

	seenRemoteIDs := make(map[string]struct{}, len(request.Items))
	items := make([]ConnectorInboundRemoteIdentityRecoveryBatchItem, len(request.Items))
	copy(items, request.Items)
	for i := range items {
		items[i].RemoteID = strings.TrimSpace(items[i].RemoteID)
		items[i].ExpectedSourceObject = strings.TrimSpace(items[i].ExpectedSourceObject)
		items[i].ExpectedSourceRecordID = strings.TrimSpace(items[i].ExpectedSourceRecordID)
		if items[i].RemoteID == "" || items[i].ExpectedSourceObject == "" || items[i].ExpectedSourceRecordID == "" {
			return request, fmt.Errorf("connector inbound remote identity recovery batch item %d requires remote ID and expected source pair", i)
		}
		if _, duplicate := seenRemoteIDs[items[i].RemoteID]; duplicate {
			return request, fmt.Errorf("connector inbound remote identity recovery batch has duplicate remote ID %q", items[i].RemoteID)
		}
		seenRemoteIDs[items[i].RemoteID] = struct{}{}
	}
	sort.Slice(items, func(i, j int) bool {
		identity := ConnectorPushDurableIdentity{
			ConnectorOwnerKey: request.Selector.ConnectorOwnerKey,
			DestinationSlotID: request.Selector.DestinationSlotID,
		}
		left := ConnectorPushRemoteLocatorHash(identity, items[i].RemoteID)
		right := ConnectorPushRemoteLocatorHash(identity, items[j].RemoteID)
		if left == right {
			return items[i].RemoteID < items[j].RemoteID
		}
		return left < right
	})
	request.Items = items
	return request, nil
}

// recoverInAmbientTx is private implementation detail of the service-owned
// transaction. Batch recovery, when added, must likewise be service-owned and
// must not expose this ambient body as a caller transaction contract.
func (s *ConnectorInboundRemoteIdentityRecoveryService) recoverInAmbientTx(
	ctx context.Context,
	request ConnectorInboundRemoteIdentityRecoveryRequest,
) (*ConnectorInboundRemoteIdentityRecoveryResult, error) {
	if !s.txManager.IsTransactionActive(ctx) {
		return nil, fmt.Errorf("connector inbound remote identity recovery requires a live pessimistic transaction")
	}
	selector, err := normalizeConnectorInboundRemoteIdentityRecoverySelector(request.Selector)
	if err != nil {
		return nil, err
	}
	hints, err := s.store.FindInboundRemoteIDEvidenceHintsInAmbientTx(ctx, selector)
	if err != nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery evidence hint: %w", err)
	}
	installed, err := s.installed.ResolveConnectorInboundRemoteIdentityRecoveryInstalledBindings(ctx, selector)
	if err != nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery installed bindings: %w", err)
	}
	bindingByKey, evidenceBindings, err := normalizeConnectorInboundRemoteIdentityRecoveryInstalledBindings(selector, installed)
	if err != nil {
		return nil, err
	}
	hint, err := selectConnectorInboundRemoteIdentityRecoveryHint(selector, hints, bindingByKey)
	if err != nil {
		return nil, err
	}
	if hint == nil {
		return nil, nil
	}
	if err := validateConnectorInboundRemoteIdentityRecoveryEvidence(selector, *hint); err != nil {
		return nil, err
	}

	binding, ok := bindingByKey[connectorInboundRemoteIDEvidenceBindingKey(ConnectorInboundRemoteIDEvidenceBinding{
		SourceObject:  hint.SourceObject,
		RemoteIDField: hint.RemoteIDField,
	})]
	if !ok {
		return nil, fmt.Errorf("%w: evidence hint is outside the locked installed recovery bindings", ErrConnectorPushDurableConflict)
	}
	validation, err := s.sourceValidator.ValidateConnectorInboundRemoteIdentityRecoverySource(ctx, *hint, binding)
	if err != nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery source validation: %w", err)
	}
	mirrorStatus, err := normalizeConnectorInboundRecoveryMirrorStatus(validation.CompatibilityMirrorStatus, false)
	if err != nil {
		return nil, err
	}
	identity := binding.Identity
	identity.SourceRecordID = hint.SourceRecordId
	linkKey, err := identity.LinkKey()
	if err != nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery durable identity: %w", err)
	}
	if err := s.store.LockConnectorRemoteIdentityLaneInAmbientTx(ctx, identity, selector.RemoteID); err != nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery admission lane: %w", err)
	}

	// The non-locking hint is not authority. Re-read the installed allowlist
	// after source and lane locks, and require the exact hinted row.
	evidence, err := s.store.FindUniqueInboundRemoteIDEvidenceForRecoveryInAmbientTx(
		ctx,
		selector,
		evidenceBindings,
	)
	if err != nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery inverse evidence: %w", err)
	}
	if evidence == nil || strings.TrimSpace(evidence.ID) != strings.TrimSpace(hint.ID) {
		return nil, fmt.Errorf("%w: authoritative recovery evidence changed after admission", ErrConnectorPushDurableConflict)
	}
	if err := validateConnectorInboundRemoteIdentityRecoveryEvidence(selector, *evidence); err != nil {
		return nil, err
	}
	finalization := ConnectorInboundRemoteIdentityFinalization{
		Evidence: *evidence,
		Adoption: ConnectorPushLinkAdoption{
			LinkKey:  linkKey,
			RemoteID: selector.RemoteID,
			Identity: identity,
			Snapshot: binding.Snapshot,
		},
		CompatibilityMirrorStatus: mirrorStatus,
	}
	if _, err := s.store.CheckConnectorRemoteIdentityRecoveryHistoryInAmbientTx(ctx, finalization); err != nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery locator history preflight: %w", err)
	}
	if evidence.EvidenceStatus == ConnectorInboundEvidenceStatusConsumed &&
		strings.TrimSpace(evidence.ConsumedByLinkKey) != linkKey {
		return nil, fmt.Errorf("%w: consumed recovery evidence does not name the exact link", ErrConnectorPushDurableConflict)
	}
	if err := s.store.FinalizeInboundRemoteIdentityInAmbientTx(ctx, finalization); err != nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery strict finalization: %w", err)
	}
	if err := s.store.AssertExactInboundRemoteIDEvidenceRecoveryInAmbientTx(
		ctx,
		selector,
		evidenceBindings,
		evidence.ID,
		linkKey,
	); err != nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery fan-in readback: %w", err)
	}
	if err := s.store.AssertExactConnectorRemoteIdentityRecoveryHistoryInAmbientTx(ctx, finalization); err != nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery locator history: %w", err)
	}
	return &ConnectorInboundRemoteIdentityRecoveryResult{
		EvidenceID:     evidence.ID,
		LinkKey:        linkKey,
		SourceObject:   evidence.SourceObject,
		SourceRecordID: evidence.SourceRecordId,
		RemoteIDField:  evidence.RemoteIDField,
	}, nil
}

func normalizeConnectorInboundRemoteIdentityRecoveryInstalledBindings(
	selector ConnectorInboundRemoteIdentityRecoverySelector,
	installed ConnectorInboundRemoteIdentityRecoveryInstalledBindings,
) (map[string]ConnectorInboundRemoteIdentityRecoveryBinding, []ConnectorInboundRemoteIDEvidenceBinding, error) {
	if strings.TrimSpace(installed.Generation) == "" {
		return nil, nil, fmt.Errorf("connector inbound remote identity recovery requires a locked installed generation")
	}
	if len(installed.Bindings) == 0 {
		return nil, nil, fmt.Errorf("connector inbound remote identity recovery requires at least one locked installed binding")
	}

	bindingByKey := make(map[string]ConnectorInboundRemoteIdentityRecoveryBinding, len(installed.Bindings))
	evidenceBindings := make([]ConnectorInboundRemoteIDEvidenceBinding, 0, len(installed.Bindings))
	for _, binding := range installed.Bindings {
		evidenceBinding, err := normalizeConnectorInboundRemoteIDEvidenceBinding(binding.EvidenceBinding)
		if err != nil {
			return nil, nil, err
		}
		key := connectorInboundRemoteIDEvidenceBindingKey(evidenceBinding)
		if _, exists := bindingByKey[key]; exists {
			return nil, nil, fmt.Errorf("connector inbound remote identity recovery has duplicate installed binding %s/%s", evidenceBinding.SourceObject, evidenceBinding.RemoteIDField)
		}

		identity := binding.Identity
		identity.ConnectorOwnerKey = strings.TrimSpace(identity.ConnectorOwnerKey)
		identity.ActionSlotID = strings.TrimSpace(identity.ActionSlotID)
		identity.ActionType = strings.TrimSpace(identity.ActionType)
		identity.SourceObject = strings.TrimSpace(identity.SourceObject)
		identity.SourceRecordID = strings.TrimSpace(identity.SourceRecordID)
		identity.DestinationSlotID = strings.TrimSpace(identity.DestinationSlotID)
		if identity.SourceRecordID != "" {
			return nil, nil, fmt.Errorf("connector inbound remote identity recovery installed identity SourceRecordID must be empty")
		}
		if identity.ConnectorOwnerKey != selector.ConnectorOwnerKey {
			return nil, nil, fmt.Errorf("connector inbound remote identity recovery installed binding owner does not match trusted owner")
		}
		if identity.DestinationSlotID != selector.DestinationSlotID {
			return nil, nil, fmt.Errorf("connector inbound remote identity recovery installed binding destination slot does not match trusted destination slot")
		}
		if identity.SourceObject != evidenceBinding.SourceObject {
			return nil, nil, fmt.Errorf("connector inbound remote identity recovery installed identity source object does not match evidence binding")
		}
		if identity.ActionSlotID == "" || identity.ActionType == "" {
			return nil, nil, fmt.Errorf("connector inbound remote identity recovery installed identity action slot and action type are required")
		}

		binding.EvidenceBinding = evidenceBinding
		binding.Identity = identity
		binding.Snapshot.RemoteIDField = strings.TrimSpace(binding.Snapshot.RemoteIDField)
		if binding.Snapshot.RemoteIDField != evidenceBinding.RemoteIDField {
			return nil, nil, fmt.Errorf("connector inbound remote identity recovery installed snapshot remote ID field does not match evidence binding")
		}
		bindingByKey[key] = binding
		evidenceBindings = append(evidenceBindings, evidenceBinding)
	}
	sort.Slice(evidenceBindings, func(i, j int) bool {
		return connectorInboundRemoteIDEvidenceBindingKey(evidenceBindings[i]) < connectorInboundRemoteIDEvidenceBindingKey(evidenceBindings[j])
	})
	return bindingByKey, evidenceBindings, nil
}

func selectConnectorInboundRemoteIdentityRecoveryHint(
	selector ConnectorInboundRemoteIdentityRecoverySelector,
	hints []ConnectorInboundRemoteIDEvidence,
	bindingByKey map[string]ConnectorInboundRemoteIdentityRecoveryBinding,
) (*ConnectorInboundRemoteIDEvidence, error) {
	var candidate *ConnectorInboundRemoteIDEvidence
	for i := range hints {
		hint := hints[i]
		key := connectorInboundRemoteIDEvidenceBindingKey(ConnectorInboundRemoteIDEvidenceBinding{
			SourceObject: strings.TrimSpace(hint.SourceObject), RemoteIDField: strings.TrimSpace(hint.RemoteIDField),
		})
		if _, ok := bindingByKey[key]; !ok {
			continue
		}
		if err := validateConnectorInboundRemoteIdentityRecoveryEvidence(selector, hint); err != nil {
			return nil, err
		}
		if candidate != nil {
			return nil, fmt.Errorf("%w: multiple authored or consumed evidence rows match locked installed recovery bindings", ErrConnectorPushDurableConflict)
		}
		copy := hint
		candidate = &copy
	}
	return candidate, nil
}

func normalizeConnectorInboundRemoteIdentityRecoverySelector(selector ConnectorInboundRemoteIdentityRecoverySelector) (ConnectorInboundRemoteIdentityRecoverySelector, error) {
	selector.ConnectorOwnerKey = strings.TrimSpace(selector.ConnectorOwnerKey)
	selector.ConnectorNamespace = strings.TrimSpace(selector.ConnectorNamespace)
	selector.DestinationSlotID = strings.TrimSpace(selector.DestinationSlotID)
	selector.RemoteID = strings.TrimSpace(selector.RemoteID)
	if selector.ConnectorOwnerKey == "" || selector.ConnectorNamespace == "" || selector.DestinationSlotID == "" || selector.RemoteID == "" {
		return selector, fmt.Errorf("connector inbound remote identity recovery owner, namespace, destination slot, and remote ID are required")
	}
	return selector, nil
}

func normalizeConnectorInboundRemoteIDEvidenceBinding(binding ConnectorInboundRemoteIDEvidenceBinding) (ConnectorInboundRemoteIDEvidenceBinding, error) {
	binding.SourceObject = strings.TrimSpace(binding.SourceObject)
	binding.RemoteIDField = strings.TrimSpace(binding.RemoteIDField)
	if binding.SourceObject == "" || binding.RemoteIDField == "" {
		return binding, fmt.Errorf("connector inbound remote identity recovery source object and remote ID field are required")
	}
	return binding, nil
}

func connectorInboundRemoteIDEvidenceBindingKey(binding ConnectorInboundRemoteIDEvidenceBinding) string {
	return binding.SourceObject + "\x1f" + binding.RemoteIDField
}

func validateConnectorInboundRemoteIdentityRecoveryEvidence(selector ConnectorInboundRemoteIdentityRecoverySelector, evidence ConnectorInboundRemoteIDEvidence) error {
	if strings.TrimSpace(evidence.ID) == "" ||
		strings.TrimSpace(evidence.SourceObject) == "" ||
		strings.TrimSpace(evidence.SourceRecordId) == "" ||
		strings.TrimSpace(evidence.RemoteIDField) == "" ||
		strings.TrimSpace(evidence.ConnectorOwnerKey) != selector.ConnectorOwnerKey ||
		strings.TrimSpace(evidence.ConnectorNamespace) != selector.ConnectorNamespace ||
		strings.TrimSpace(evidence.RemoteId) != selector.RemoteID {
		return fmt.Errorf("%w: inverse evidence does not match the recovery selector", ErrConnectorPushDurableConflict)
	}
	switch strings.TrimSpace(evidence.EvidenceStatus) {
	case ConnectorInboundEvidenceStatusAuthored:
		expectedActiveKey := ConnectorInboundEvidenceActiveKey(
			evidence.ConnectorOwnerKey,
			evidence.SourceObject,
			evidence.SourceRecordId,
			evidence.RemoteIDField,
		)
		if strings.TrimSpace(evidence.ActiveKey) != expectedActiveKey || strings.TrimSpace(evidence.ConsumedByLinkKey) != "" {
			return fmt.Errorf("%w: inverse authored evidence has invalid active state", ErrConnectorPushDurableConflict)
		}
	case ConnectorInboundEvidenceStatusConsumed:
		if strings.TrimSpace(evidence.ActiveKey) != "" || strings.TrimSpace(evidence.ConsumedByLinkKey) == "" {
			return fmt.Errorf("%w: inverse consumed evidence has invalid link state", ErrConnectorPushDurableConflict)
		}
	default:
		return fmt.Errorf("%w: inverse evidence has unsupported status", ErrConnectorPushDurableConflict)
	}
	return nil
}

func normalizeConnectorInboundRecoveryMirrorStatus(status ConnectorPushCompatibilityMirrorStatus, allowDefault bool) (ConnectorPushCompatibilityMirrorStatus, error) {
	status = ConnectorPushCompatibilityMirrorStatus(strings.TrimSpace(string(status)))
	if status == "" && allowDefault {
		return ConnectorPushCompatibilityMirrorStatus(ConnectorPushCompatibilityMirrorSynced), nil
	}
	switch string(status) {
	case ConnectorPushCompatibilityMirrorSynced, ConnectorPushCompatibilityMirrorDrifted:
		return status, nil
	default:
		return "", fmt.Errorf("connector inbound remote identity recovery mirror status must be synced or drifted")
	}
}