package records

import (
	"context"
	stderrors "errors"
	"fmt"
	"strings"
	"time"

	"github.com/nexuscrm/kernel/pkg/apierr"
	"github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/kernel/pkg/telemetry"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	sharedErrors "github.com/nexuscrm/shared/pkg/errors"
	"github.com/nexuscrm/shared/pkg/models"
	zlog "github.com/rs/zerolog/log"
)

func recordMutationSuccess(id string) contract.RecordMutationResult {
	return contract.RecordMutationResult{
		ID:      id,
		Success: true,
		Errors:  []contract.RecordMutationError{},
	}
}

func recordMutationUpsertSuccess(id string, created bool) contract.RecordMutationResult {
	return contract.RecordMutationResult{
		ID:      id,
		Success: true,
		Created: &created,
		Errors:  []contract.RecordMutationError{},
	}
}

func recordMutationError(id string, err contract.RecordMutationError) contract.RecordMutationResult {
	return recordMutationErrors(id, []contract.RecordMutationError{err})
}

func recordMutationErrors(id string, errs []contract.RecordMutationError) contract.RecordMutationResult {
	normalized := append([]contract.RecordMutationError(nil), errs...)
	for i := range normalized {
		if normalized[i].Fields == nil {
			normalized[i].Fields = []string{}
		}
	}
	return contract.RecordMutationResult{
		ID:      id,
		Success: false,
		Errors:  normalized,
	}
}

func recordMutationResultIsEmpty(result contract.RecordMutationResult) bool {
	return result.ID == "" && !result.Success && result.Created == nil && len(result.Errors) == 0
}

// allOrNoneRolledBackEntry is the neutral per-record marker for records that
// would have succeeded but were rolled back due to allOrNone=true. The engine is
// app-blind: it emits only the neutral Kind and leaves Message empty so the wire
// dialect renders the product's fixed English (the Salesforce v63 dialect renders
// the "Record rolled back because not all records were valid..." string from this
// Kind). This is the sole constructor for the rollback marker; markAllOrNoneRollback
// routes every rewritten slot through it.
func allOrNoneRolledBackEntry(id string) contract.RecordMutationResult {
	return recordMutationError(id, contract.RecordMutationError{
		Kind:    contract.RecordMutationErrorAllOrNoneRolledBack,
		Message: "",
		Fields:  []string{},
	})
}

func processingHaltedEntry(recordID string, message string) contract.RecordMutationResult {
	return recordMutationError(recordID, contract.RecordMutationError{
		Kind:    contract.RecordMutationErrorProcessingHalted,
		Message: message,
		Fields:  []string{},
	})
}

func internalRecordMutationEntry(recordID string) contract.RecordMutationResult {
	return recordMutationError(recordID, contract.RecordMutationError{
		Kind:    contract.RecordMutationErrorInternal,
		Message: sanitizedInternalErrorMessage,
		Fields:  []string{},
	})
}

func invalidIDEntry(recordID string, message string) contract.RecordMutationResult {
	return recordMutationError(recordID, contract.RecordMutationError{
		Kind:    contract.RecordMutationErrorInvalidID,
		Message: message,
		Fields:  []string{},
	})
}

func malformedIDEntry(recordID string, message string) contract.RecordMutationResult {
	return recordMutationError(recordID, contract.RecordMutationError{
		Kind:    contract.RecordMutationErrorMalformedID,
		Message: message,
		Fields:  []string{},
	})
}

func notFoundEntry(recordID string, message string) contract.RecordMutationResult {
	return recordMutationError(recordID, contract.RecordMutationError{
		Kind:    contract.RecordMutationErrorNotFound,
		Message: message,
		Fields:  []string{},
	})
}

func recordMutationErrorKindFromDomainKind(kind errors.WireConditionKind) contract.RecordMutationErrorKind {
	switch kind {
	case errors.WireInvalidReference:
		return contract.RecordMutationErrorInvalidCrossReference
	case errors.WireFieldIntegrity:
		return contract.RecordMutationErrorFieldIntegrity
	case errors.WireInvalidOperation:
		return contract.RecordMutationErrorInvalidOperation
	case errors.WireInvalidField:
		return contract.RecordMutationErrorInvalidField
	case errors.WireNotFound:
		return contract.RecordMutationErrorNotFound
	default:
		return contract.RecordMutationErrorProcessingHalted
	}
}

func recordsFromMap(recordMap map[string]models.SObject) []models.SObject {
	records := make([]models.SObject, 0, len(recordMap))
	for _, record := range recordMap {
		records = append(records, record)
	}
	return records
}

func stripRecordMetadataFields(record models.SObject) models.SObject {
	filtered := make(models.SObject, len(record))
	for field, value := range record {
		if models.IsRecordMetadataKey(field) {
			continue
		}
		filtered[field] = value
	}
	return filtered
}

func (ps *PersistenceService) isReadableUserField(ctx context.Context, objectName, field string, user *models.UserSession) (bool, error) {
	if models.IsRecordMetadataKey(field) {
		return false, nil
	}
	if ps == nil || ps.permissions == nil {
		return true, nil
	}
	return ps.permissions.CheckFieldVisibilityWithUser(ctx, objectName, field, user)
}

func (ps *PersistenceService) isReadableUserFieldWithSchema(ctx context.Context, schema *models.ObjectMetadata, fieldName string, user *models.UserSession) (bool, error) {
	if models.IsRecordMetadataKey(fieldName) {
		return false, nil
	}
	if ps == nil || ps.permissions == nil {
		return true, nil
	}
	field := models.FindField(schema, fieldName)
	if field == nil {
		return ps.permissions.CheckFieldVisibilityWithUser(ctx, schema.APIName, fieldName, user)
	}
	return ps.permissions.CheckFieldVisibilityWithSchema(ctx, schema, field, user)
}

func (ps *PersistenceService) filterReadBlockedFields(ctx context.Context, objectName string, record models.SObject) models.SObject {
	if ps == nil || ps.permissions == nil {
		return stripRecordMetadataFields(record)
	}

	filtered := make(models.SObject, len(record))
	for field, value := range record {
		if models.IsRecordMetadataKey(field) {
			continue
		}
		if !ps.permissions.IsFieldReadBlocked(ctx, objectName, field) {
			filtered[field] = value
		}
	}
	schemautil.CopyExactDecimalsForPresentFields(record, filtered)
	return filtered
}

// ResolveObjectTypes resolves record IDs to their physical object API names in
// one repository call. It is the batch counterpart to GetRecordType for public
// endpoints that accept up to 200 IDs and must not fan out into one metadata
// scan per record.
func (ps *PersistenceService) ResolveObjectTypes(ctx context.Context, ids []string) (map[string]string, error) {
	if len(ids) == 0 {
		return map[string]string{}, nil
	}
	if ps == nil || ps.repo == nil {
		return nil, errors.NewInternalError("record object type resolver is not configured", nil)
	}

	idTypeMap, err := ps.repo.FindObjectTypesForIDs(ctx, ids, nil)
	if err != nil {
		return nil, errors.NewInternalError("failed to discover object types", err)
	}
	return idTypeMap, nil
}

type recordObjectTypeCandidateRepository interface {
	FindObjectTypeCandidatesForIDs(ctx context.Context, ids []string) (map[string][]string, error)
}

// ResolveObjectTypeCandidates returns every logical object currently owning an
// opaque record ID. Repositories without candidate discovery fail closed rather
// than silently reverting to first-match resolution.
func (ps *PersistenceService) ResolveObjectTypeCandidates(ctx context.Context, ids []string) (map[string][]string, error) {
	candidateIDs := make([]string, 0, len(ids))
	seen := make(map[string]struct{}, len(ids))
	for _, id := range ids {
		if id == "" {
			continue
		}
		if _, exists := seen[id]; exists {
			continue
		}
		seen[id] = struct{}{}
		candidateIDs = append(candidateIDs, id)
	}
	if len(candidateIDs) == 0 {
		return map[string][]string{}, nil
	}
	if ps == nil || ps.repo == nil {
		return nil, errors.NewInternalError("record object type resolver is not configured", nil)
	}
	resolver, ok := ps.repo.(recordObjectTypeCandidateRepository)
	if !ok {
		return nil, errors.NewInternalError("record repository does not support object type candidate discovery", nil)
	}
	resolved, err := resolver.FindObjectTypeCandidatesForIDs(ctx, candidateIDs)
	if err != nil {
		return nil, errors.NewInternalError("failed to discover object type candidates", err)
	}
	return resolved, nil
}

// markAllOrNoneRollback rewrites every record result into the SF-verbatim
// allOrNone-rollback shape:
//   - Previously success=true → overwritten with ALL_OR_NONE_OPERATION_ROLLED_BACK
//     (the record would have committed but didn't due to a sibling's failure)
//   - Previously nil (loop short-circuited before reaching this slot) → also
//     ALL_OR_NONE_OPERATION_ROLLED_BACK so JSON contains a concrete entry for
//     every input rather than null. SF v63.0 includes every input record in
//     the response.
//   - Previously populated failure → kept intact (the failing record keeps
//     its specific errorCode such as INVALID_FIELD_FOR_INSERT_UPDATE).
//
// idFor resolves the record-id for positions never processed (where results[i]
// is nil) so the caller can recover the id from its input slice.
func markAllOrNoneRollback(results []contract.RecordMutationResult, idFor func(i int) string) {
	for i := range results {
		if recordMutationResultIsEmpty(results[i]) {
			id := ""
			if idFor != nil {
				id = idFor(i)
			}
			results[i] = allOrNoneRolledBackEntry(id)
			continue
		}
		if results[i].Success {
			results[i] = allOrNoneRolledBackEntry(results[i].ID)
		}
	}
}

func groupMembershipClosureObjectNamesFromSObjects(records []models.SObject) []string {
	objectNames := make([]string, 0, len(records))
	for _, rec := range records {
		if objectName := models.RecordObjectType(rec); objectName != "" {
			objectNames = append(objectNames, objectName)
		}
	}
	return objectNames
}

type deleteRecordTargetResolution struct {
	objectName string
	err        error
	missing    bool
	ambiguous  bool
}

func (ps *PersistenceService) resolveDeleteRecordTargetsBeforeMutation(ctx context.Context, ids []string) (map[string]deleteRecordTargetResolution, []string, error) {
	targetsByID := make(map[string]deleteRecordTargetResolution, len(ids))
	objectNames := make([]string, 0, len(ids))
	seen := make(map[string]struct{}, len(ids))
	for _, id := range ids {
		if id == "" {
			continue
		}
		if _, ok := targetsByID[id]; ok {
			continue
		}
		objectName, err := ps.repo.GetRecordType(ctx, id)
		if errors.IsInternal(err) {
			return nil, nil, err
		}
		if err != nil {
			targetsByID[id] = deleteRecordTargetResolution{
				err:       err,
				missing:   errors.IsNotFound(err),
				ambiguous: !errors.IsNotFound(err),
			}
			continue
		}
		if objectName == "" {
			targetsByID[id] = deleteRecordTargetResolution{
				missing:   true,
				ambiguous: true,
			}
			continue
		}
		targetsByID[id] = deleteRecordTargetResolution{objectName: objectName}
		for _, deleteObjectName := range ps.deleteTransactionObjectNames(ctx, objectName) {
			key := strings.ToLower(strings.TrimSpace(deleteObjectName))
			if key == "" {
				continue
			}
			if _, ok := seen[key]; ok {
				continue
			}
			seen[key] = struct{}{}
			objectNames = append(objectNames, deleteObjectName)
		}
	}
	return targetsByID, objectNames, nil
}

const sanitizedInternalErrorMessage = "An internal server error occurred. Contact your system administrator."

func bulkErrorEntry(recordID string, err error) contract.RecordMutationResult {
	kind := contract.RecordMutationErrorProcessingHalted
	fieldsList := []string{}
	message := err.Error()
	if errors.IsInternal(err) {
		return internalRecordMutationEntry(recordID)
	}
	// Neutral engine-condition errors are preserved as semantic mutation error
	// kinds here; v63 maps them to its statusCode strings at the boundary.
	var domainErr *errors.DomainWireError
	if stderrors.As(err, &domainErr) {
		return recordMutationError(recordID, contract.RecordMutationError{
			Kind:                  recordMutationErrorKindFromDomainKind(domainErr.Kind),
			Message:               domainErr.Message,
			Fields:                append([]string(nil), domainErr.Fields...),
			CreateFromSourceGuard: domainErr.CreateFromSourceGuard.Clone(),
		})
	}
	if validationErrors, ok := errors.AsClientValidationErrors(err); ok {
		mutationErrors := make([]contract.RecordMutationError, 0, len(validationErrors))
		for _, validationErr := range validationErrors {
			if failures, ok := errors.AsValidationRuleFailures(validationErr); ok {
				for _, failure := range failures {
					fields := []string{}
					if field := strings.TrimSpace(failure.Field); field != "" {
						fields = append(fields, field)
					}
					mutationErrors = append(mutationErrors, contract.RecordMutationError{
						Kind:    contract.RecordMutationErrorCustomValidation,
						Message: failure.Message,
						Fields:  fields,
					})
				}
				continue
			}
			if deniedFields, deniedMessage, ok := sharedErrors.FieldWriteDeniedDetails(validationErr); ok {
				mutationErrors = append(mutationErrors, contract.RecordMutationError{
					Kind:    contract.RecordMutationErrorInvalidFieldForUpdate,
					Message: deniedMessage,
					Fields:  append([]string(nil), deniedFields...),
				})
				continue
			}
			if fields, invalidMessage, ok := sharedErrors.InvalidFieldForWriteDetails(validationErr); ok {
				mutationErrors = append(mutationErrors, contract.RecordMutationError{
					Kind:    contract.RecordMutationErrorInvalidFieldForUpdate,
					Message: invalidMessage,
					Fields:  append([]string(nil), fields...),
				})
				continue
			}
			if field, _, ok := sharedErrors.ValidationDetails(validationErr); ok {
				fields := []string{}
				validationKind := contract.RecordMutationErrorProcessingHalted
				if field = strings.TrimSpace(field); field != "" {
					fields = append(fields, field)
					validationKind = contract.RecordMutationErrorFieldIntegrity
				}
				mutationErrors = append(mutationErrors, contract.RecordMutationError{
					Kind:    validationKind,
					Message: validationErr.Error(),
					Fields:  fields,
				})
			}
		}
		if len(mutationErrors) > 0 {
			return recordMutationErrors(recordID, mutationErrors)
		}
	}
	return recordMutationError(recordID, contract.RecordMutationError{
		Kind:    kind,
		Message: message,
		Fields:  fieldsList,
	})
}

// CreateRecords creates multiple records, optionally in a single transaction (allOrNone)
func (ps *PersistenceService) CreateRecords(ctx context.Context, records []models.SObject, allOrNone bool, user *models.UserSession) ([]contract.RecordMutationResult, error) {
	if len(records) == 0 {
		return nil, nil
	}

	// Validate all records have type before proceeding
	for i, rec := range records {
		objType := models.RecordObjectType(rec)
		if objType == "" {
			return nil, apierr.Invalid(fmt.Sprintf("record %d missing attributes.type", i))
		}
	}

	results := make([]contract.RecordMutationResult, len(records))
	objectNames := groupMembershipClosureObjectNamesFromSObjects(records)
	if err := ps.rejectExclusiveSiblingSharedWriteSurface(ctx, fmt.Sprintf("CreateRecords(allOrNone=%t)", allOrNone), objectNames); err != nil {
		return nil, err
	}

	processFn := func(txCtx context.Context) error {
		for i, rec := range records {
			objType := models.RecordObjectType(rec)
			if err := rejectGenericListViewMutation(objType, "collection create"); err != nil {
				results[i] = bulkErrorEntry("", err)
				if allOrNone {
					return err
				}
				continue
			}

			data := make(map[string]interface{})
			for k, v := range rec {
				if !models.IsRecordMetadataKey(k) {
					data[k] = v
				}
			}

			// Normalize field names for Salesforce compatibility (case-insensitive)
			data = ps.normalizeFieldNamesForObject(txCtx, objType, data)

			// Call Insert from persistence_insert.go
			insertResult, err := ps.Insert(txCtx, objType, data, user)
			if err != nil {
				// Populate per-record entry BEFORE returning rollback signal
				// so SF-parity reporting survives allOrNone=true.
				results[i] = bulkErrorEntry("", err)
				if allOrNone {
					return err
				}
				continue
			}

			// insertResult is models.SObject (map[string]interface{})
			var idStr string
			if val, ok := insertResult["id"].(string); ok {
				idStr = val
			} else if val, ok := insertResult["Id"].(string); ok {
				idStr = val
			}

			if idStr == "" {
				// ID not found in result; proceed with empty string.
			}

			results[i] = recordMutationSuccess(idStr)
		}
		return nil
	}

	if allOrNone {
		err := ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, objectNames, processFn)
		if err != nil {
			// Infrastructure failures (tx begin/commit/rollback, IO) must
			// surface — do not mask them as per-record rollbacks.
			if errors.IsInternal(err) {
				return nil, err
			}
			// SF parity: allOrNone=true rollback → HTTP 200 + per-record
			// failure (ALL_OR_NONE_OPERATION_ROLLED_BACK for prior-success).
			// New records have no pre-assigned id, so nil-id is correct.
			markAllOrNoneRollback(results, func(int) string { return "" })
			return results, nil
		}
	} else {
		_ = processFn(ctx)
	}

	return results, nil
}

// normalizeFieldNamesForObject converts field names to match schema (case-insensitive)
// Salesforce API is case-insensitive but NexusCRM stores fields in lowercase
func (ps *PersistenceService) normalizeFieldNamesForObject(ctx context.Context, objectName string, data map[string]interface{}) map[string]interface{} {
	schema := ps.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return data // If no schema, return as-is and let validation fail later
	}

	// Build lowercase-to-actual field name map
	fieldNameMap := make(map[string]string)
	for _, f := range schema.Fields {
		fieldNameMap[strings.ToLower(f.APIName)] = f.APIName
	}

	// Metadata fields to strip (not actual data fields)
	metadataFields := map[string]bool{
		"attributes":                 true, // Salesforce client libs add {"type": "ObjectName"}
		models.ObjectTypeMetadataKey: true,
	}

	// Normalize field names in data
	normalized := make(map[string]interface{})
	for key, value := range data {
		// Skip metadata fields
		if metadataFields[strings.ToLower(key)] {
			continue
		}
		if actualName, ok := fieldNameMap[strings.ToLower(key)]; ok {
			normalized[actualName] = value
		} else {
			// Keep original name for validation error
			normalized[key] = value
		}
	}
	return normalized
}

// UpdateRecords updates multiple records
func (ps *PersistenceService) UpdateRecords(ctx context.Context, records []models.SObject, allOrNone bool, user *models.UserSession) ([]contract.RecordMutationResult, error) {
	results := make([]contract.RecordMutationResult, len(records))
	objectNames := groupMembershipClosureObjectNamesFromSObjects(records)
	if err := ps.rejectExclusiveSiblingSharedWriteSurface(ctx, fmt.Sprintf("UpdateRecords(allOrNone=%t)", allOrNone), objectNames); err != nil {
		return nil, err
	}

	processFn := func(txCtx context.Context) error {
		for i, rec := range records {
			objType := models.RecordObjectType(rec)

			id, _ := rec["Id"].(string)
			if id == "" {
				id, _ = rec["id"].(string)
			}

			if id == "" || objType == "" {
				errMsg := fmt.Sprintf("missing id or type at index %d", i)
				results[i] = malformedIDEntry("", errMsg)
				if allOrNone {
					return errors.NewValidationError("", errMsg)
				}
				continue
			}
			if err := rejectGenericListViewMutation(objType, "collection update"); err != nil {
				results[i] = bulkErrorEntry(id, err)
				if allOrNone {
					return err
				}
				continue
			}

			data := make(map[string]interface{})
			for k, v := range rec {
				if !models.IsRecordMetadataKey(k) && k != "Id" && k != "id" {
					data[k] = v
				}
			}

			err := ps.Update(txCtx, objType, id, data, user)
			if err != nil {
				// Populate per-record entry BEFORE returning the rollback
				// signal so SF-parity reporting survives allOrNone=true.
				results[i] = bulkErrorEntry(id, err)
				if allOrNone {
					return err
				}
				continue
			}
			results[i] = recordMutationSuccess(id)
		}
		return nil
	}

	if allOrNone {
		err := ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, objectNames, processFn)
		if err != nil {
			// Infrastructure failures (tx begin/commit/rollback, IO) must
			// surface — do not mask them as per-record rollbacks.
			if errors.IsInternal(err) {
				return nil, err
			}
			// SF parity: allOrNone=true batch failure returns HTTP 200 with
			// per-record results — the record that caused the rollback keeps
			// its specific errorCode (e.g. INVALID_FIELD_FOR_INSERT_UPDATE),
			// prior records flip to ALL_OR_NONE_OPERATION_ROLLED_BACK. Records
			// the loop never reached recover their id from the input slice.
			markAllOrNoneRollback(results, func(i int) string {
				if id, ok := records[i]["Id"].(string); ok && id != "" {
					return id
				}
				if id, ok := records[i]["id"].(string); ok {
					return id
				}
				return ""
			})
			return results, nil
		}
	} else {
		_ = processFn(ctx)
	}

	return results, nil
}

// DeleteRecords deletes multiple records
// This implementation looks up the object type for each ID from the record_types table
func (ps *PersistenceService) DeleteRecords(ctx context.Context, ids []string, allOrNone bool, user *models.UserSession) ([]contract.RecordMutationResult, error) {
	if len(ids) == 0 {
		return nil, nil
	}

	results := make([]contract.RecordMutationResult, len(ids))
	targetsByID, objectNames, err := ps.resolveDeleteRecordTargetsBeforeMutation(ctx, ids)
	if err != nil {
		return nil, err
	}
	if allOrNone {
		if err := ps.rejectExclusiveSiblingDeleteRecordsAllOrNone(ctx, objectNames); err != nil {
			return nil, err
		}
	}

	// The object type map is resolved once before any mutation so collection
	// deletes keep allOrNone rollback behavior independent from per-record type
	// discovery.
	processFn := func(txCtx context.Context) error {
		for i, id := range ids {
			// Validate ID format before querying
			if id == "" {
				errMsg := fmt.Sprintf("empty id at index %d", i)
				results[i] = invalidIDEntry(id, errMsg)
				if allOrNone {
					return errors.NewValidationError("id", errMsg)
				}
				continue
			}

			target := targetsByID[id]
			if target.objectName == "" {
				errMsg := fmt.Sprintf("could not determine object type for id %s", id)
				if target.missing {
					errMsg = fmt.Sprintf("record %s not found", id)
					results[i] = notFoundEntry(id, errMsg)
				} else {
					results[i] = processingHaltedEntry(id, errMsg)
				}
				if allOrNone {
					if target.missing {
						return errors.NewNotFoundError("record", id)
					}
					return target.err
				}
				continue
			}
			if !constants.GetSecurityContext(txCtx).BypassesCRUDandFLS() {
				if err := rejectGenericListViewMutation(target.objectName, "collection delete"); err != nil {
					results[i] = bulkErrorEntry(id, err)
					if allOrNone {
						return err
					}
					continue
				}
			}

			// Delete the record using existing Delete method
			err := ps.Delete(txCtx, target.objectName, id, user)
			if err != nil {
				results[i] = bulkErrorEntry(id, err)
				if allOrNone {
					return err
				}
				continue
			}

			results[i] = recordMutationSuccess(id)
		}
		return nil
	}

	if allOrNone {
		err := ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, objectNames, processFn)
		if err != nil {
			// Infrastructure failures (tx begin/commit/rollback, IO) must
			// surface — do not mask them as per-record rollbacks.
			if errors.IsInternal(err) {
				return nil, err
			}
			// SF parity: allOrNone=true rollback → HTTP 200 + per-record
			// failure (ALL_OR_NONE_OPERATION_ROLLED_BACK for prior-success).
			// Verified against sandbox DELETE /composite/sobjects?allOrNone=true.
			// IDs for never-processed slots come from the input slice.
			markAllOrNoneRollback(results, func(i int) string { return ids[i] })
			return results, nil
		}
	} else {
		_ = processFn(ctx)
	}

	return results, nil
}

// parentInfo carries a freshly created parent record's type, id, and the
// relationship name a child was reached through, so the tree-create engine can
// resolve and stamp the child's foreign key. It is used by the neutral
// insertTreeNodeIR path (see persistence_tree_nodes.go).
type parentInfo struct {
	Type    string
	ID      string
	RelName string
}

// resolveFK uses metadata to find the Foreign Key field name for a relationship
// It queries the Relationship metadata to find the correct FK field
func (ps *PersistenceService) resolveFK(ctx context.Context, parentType, childType, relName string) string {
	// Get all relationships where the child object has a lookup to parent
	relationships := ps.metadata.GetRelationships(ctx, childType)

	for _, rel := range relationships {
		// Match by relationship name (e.g., "Contacts" -> AccountId on Contact)
		if rel.RelationshipName == relName && rel.ParentObjectAPIName == parentType {
			return rel.FieldAPIName
		}
	}

	// Fallback: Check if the parent has a relationship defined
	// (Some relationships are defined from parent perspective)
	parentRels := ps.metadata.GetRelationships(ctx, parentType)
	for _, rel := range parentRels {
		if rel.RelationshipName == relName && rel.ChildObjectAPIName == childType {
			return rel.FieldAPIName
		}
	}

	// No backward compatibility fallback - metadata is SSoT
	// If relationship not found in metadata, return empty (caller must handle)
	return ""
}

// RetrieveMany retrieves multiple records by IDs with full permission checks.
// Enforces object-level permissions, record-level access (RLS), and FLS for each record.
// Records the user cannot access are silently omitted (not errored).
func (ps *PersistenceService) RetrieveMany(ctx context.Context, objectName string, ids []string, user *models.UserSession) (map[string]models.SObject, error) {
	if user == nil {
		return nil, apierr.Forbidden("authentication required")
	}
	if len(ids) == 0 {
		return map[string]models.SObject{}, nil
	}

	// Object-level permission check (one check for the whole batch)
	scopedRead := constants.IsScopedInternalAccess(ctx, objectName, constants.PermRead)
	privilegedRead := constants.GetSecurityContext(ctx) >= constants.SystemContext || scopedRead
	if ps.permissions != nil {
		if privilegedRead {
			if ps.permissions.IsBlockedObject(ctx, objectName, constants.PermRead) {
				return nil, apierr.Forbidden(fmt.Sprintf("blocked: %s is not accessible via generic %s, even in system context", objectName, constants.PermRead))
			}
		} else {
			if err := ps.permissions.CheckPermissionOrErrorWithUser(ctx, objectName, constants.PermRead, user); err != nil {
				return nil, err
			}
		}
	}

	var schema *models.ObjectMetadata
	if ps.metadata != nil {
		schema = ps.metadata.GetSchema(ctx, objectName)
	}

	// Batch fetch raw records
	rawRecords, err := ps.repo.FindMany(ctx, objectName, ids)
	if err != nil {
		return nil, errors.NewInternalError("failed to retrieve records", err)
	}

	// Filter by record-level access and apply FLS.
	bypassFLS := constants.GetSecurityContext(ctx).BypassesCRUDandFLS() ||
		constants.ScopedInternalAccessBypassesFLS(ctx, objectName, constants.PermRead)
	result := make(map[string]models.SObject, len(rawRecords))
	if !scopedRead && schema != nil && ps.permissions != nil {
		ps.permissions.PrimeReadAccessInputCache(ctx, schema.APIName, recordsFromMap(rawRecords), user)
	}
	for id, record := range rawRecords {
		if ps.permissions != nil {
			// Record-level access check (RLS)
			if !scopedRead && schema != nil && !ps.permissions.CheckRecordAccess(ctx, schema, record, constants.PermRead, user) {
				continue // silently omit unauthorized records
			}
			// FLS filtering (bypassed at ElevatedContext+)
			if bypassFLS {
				result[id] = ps.filterReadBlockedFields(ctx, objectName, record)
			} else {
				filtered := make(models.SObject)
				for k, v := range record {
					readable, err := ps.isReadableUserField(ctx, objectName, k, user)
					if err != nil {
						return nil, err
					}
					if readable {
						filtered[k] = v
					}
				}
				schemautil.CopyExactDecimalsForPresentFields(record, filtered)
				result[id] = filtered
			}
		} else {
			result[id] = stripRecordMetadataFields(record)
		}
	}

	records := make([]models.SObject, 0, len(result))
	for _, r := range result {
		records = append(records, r)
	}
	if err := ps.projectPermissionGrants(ctx, objectName, records); err != nil {
		return nil, err
	}

	return result, nil
}

// Retrieve retrieves a single record by ID with strict permission checks
func (ps *PersistenceService) Retrieve(ctx context.Context, objectName string, id string, user *models.UserSession) (models.SObject, error) {
	reqID := requestIDFromCtx(ctx)
	zlog.Debug().Str("request_id", reqID).Str("object", objectName).Str("record_id", id).Str("user", getUserID(user)).Msg("PersistenceService.Retrieve")

	// 1. Check Object Read Permission
	schema, err := ps.prepareOperation(ctx, objectName, constants.PermRead, user)
	if err != nil {
		return nil, err
	}

	// 2. Fetch Record
	// Uses Repo directly (bypassing QuerySvc filters, so we must manually secure it)
	record, err := ps.repo.FindOne(ctx, objectName, id)
	if err != nil {
		zlog.Error().Err(err).Str("request_id", reqID).Str("object", objectName).Str("record_id", id).Msg("PersistenceService.Retrieve repo.FindOne failed")
		return nil, errors.NewInternalError("failed to retrieve record", err)
	}
	if record == nil {
		return nil, apierr.NotFound(fmt.Sprintf("%s with ID '%s' not found", objectName, id))
	}

	// 3. Check Record Access (Sharing Rules)
	// Return 404 (not 403) to prevent leaking record existence via ID enumeration.
	scopedRead := constants.IsScopedInternalAccess(ctx, objectName, constants.PermRead)
	if !scopedRead && !ps.permissions.CheckRecordAccess(ctx, schema, record, constants.PermRead, user) {
		return nil, apierr.NotFound(fmt.Sprintf("%s with ID '%s' not found", objectName, id))
	}

	// 4. FLS (Field Level Security) — filter fields user cannot see.
	// ElevatedContext+ bypasses FLS, matching the write-path pattern in filterEditableFields.
	if constants.GetSecurityContext(ctx).BypassesCRUDandFLS() ||
		constants.ScopedInternalAccessBypassesFLS(ctx, objectName, constants.PermRead) {
		return ps.filterReadBlockedFields(ctx, objectName, record), nil
	}

	filtered := make(models.SObject)
	for k, v := range record {
		readable, err := ps.isReadableUserField(ctx, objectName, k, user)
		if err != nil {
			return nil, err
		}
		if readable {
			filtered[k] = v
		}
	}
	schemautil.CopyExactDecimalsForPresentFields(record, filtered)

	if err := ps.projectPermissionGrants(ctx, objectName, []models.SObject{filtered}); err != nil {
		zlog.Error().Err(err).Str("request_id", reqID).Str("object", objectName).Str("record_id", id).Msg("PersistenceService.Retrieve projectPermissionGrants failed")
		return nil, err
	}

	return filtered, nil
}

// RetrieveWithAccess retrieves a record and computes record-level access (Read/Edit/Delete)
// on the raw (unfiltered) record BEFORE applying FLS. This ensures criteria-based sharing
// rules evaluate against complete field values, not FLS-filtered ones.
func (ps *PersistenceService) RetrieveWithAccess(ctx context.Context, objectName string, id string, user *models.UserSession) (models.SObject, *permissions.RecordAccessResult, error) {
	reqID := requestIDFromCtx(ctx)
	zlog.Debug().Str("request_id", reqID).Str("object", objectName).Str("record_id", id).Str("user", getUserID(user)).Msg("PersistenceService.RetrieveWithAccess")
	totalStart := time.Now()
	var gateElapsed, findElapsed, readAccessElapsed, allAccessElapsed, flsElapsed, grantsElapsed time.Duration
	var gateSQLTime, gateSQLMax, findSQLTime, findSQLMax, readAccessSQLTime, readAccessSQLMax time.Duration
	var allAccessSQLTime, allAccessSQLMax, flsSQLTime, flsSQLMax, grantsSQLTime, grantsSQLMax time.Duration
	var gateSQL, findSQL, readAccessSQL, allAccessSQL, flsSQL, grantsSQL uint64
	defer func() {
		zlog.Debug().
			Str("event", "retrieve_with_access_phase_profile").
			Str("object", objectName).
			Uint64("object_gate_sql", gateSQL).
			Int64("object_gate_sql_elapsed_ms", gateSQLTime.Milliseconds()).
			Int64("object_gate_sql_max_ms", gateSQLMax.Milliseconds()).
			Int64("object_gate_elapsed_ms", gateElapsed.Milliseconds()).
			Uint64("find_one_sql", findSQL).
			Int64("find_one_sql_elapsed_ms", findSQLTime.Milliseconds()).
			Int64("find_one_sql_max_ms", findSQLMax.Milliseconds()).
			Int64("find_one_elapsed_ms", findElapsed.Milliseconds()).
			Uint64("read_access_sql", readAccessSQL).
			Int64("read_access_sql_elapsed_ms", readAccessSQLTime.Milliseconds()).
			Int64("read_access_sql_max_ms", readAccessSQLMax.Milliseconds()).
			Int64("read_access_elapsed_ms", readAccessElapsed.Milliseconds()).
			Uint64("all_access_sql", allAccessSQL).
			Int64("all_access_sql_elapsed_ms", allAccessSQLTime.Milliseconds()).
			Int64("all_access_sql_max_ms", allAccessSQLMax.Milliseconds()).
			Int64("all_access_elapsed_ms", allAccessElapsed.Milliseconds()).
			Uint64("fls_sql", flsSQL).
			Int64("fls_sql_elapsed_ms", flsSQLTime.Milliseconds()).
			Int64("fls_sql_max_ms", flsSQLMax.Milliseconds()).
			Int64("fls_elapsed_ms", flsElapsed.Milliseconds()).
			Uint64("grant_projection_sql", grantsSQL).
			Int64("grant_projection_sql_elapsed_ms", grantsSQLTime.Milliseconds()).
			Int64("grant_projection_sql_max_ms", grantsSQLMax.Milliseconds()).
			Int64("grant_projection_elapsed_ms", grantsElapsed.Milliseconds()).
			Int64("total_elapsed_ms", time.Since(totalStart).Milliseconds()).
			Msg("RetrieveWithAccess phase profile")
	}()

	// 1. Check Object Read Permission
	gateCtx, gateCounter := telemetry.WithSQLStatementCounter(ctx)
	gateStart := time.Now()
	schema, err := ps.prepareOperation(gateCtx, objectName, constants.PermRead, user)
	gateElapsed, gateSQL = time.Since(gateStart), gateCounter.Count()
	gateSQLTime, gateSQLMax = gateCounter.TotalDuration(), gateCounter.MaxDuration()
	if err != nil {
		return nil, nil, err
	}

	// 2. Fetch Record (raw, unfiltered)
	findCtx, findCounter := telemetry.WithSQLStatementCounter(ctx)
	findStart := time.Now()
	record, err := ps.repo.FindOne(findCtx, objectName, id)
	findElapsed, findSQL = time.Since(findStart), findCounter.Count()
	findSQLTime, findSQLMax = findCounter.TotalDuration(), findCounter.MaxDuration()
	if err != nil {
		zlog.Error().Err(err).Str("request_id", reqID).Str("object", objectName).Str("record_id", id).Msg("PersistenceService.RetrieveWithAccess repo.FindOne failed")
		return nil, nil, errors.NewInternalError("failed to retrieve record", err)
	}
	if record == nil {
		return nil, nil, apierr.NotFound(fmt.Sprintf("%s with ID '%s' not found", objectName, id))
	}

	// 3. Prime the complete record-access inputs for this exact candidate. The
	// batch repositories return authored access levels, not a pre-decided Read
	// boolean, so the same snapshot is valid for Read/Edit/Delete below. If a
	// repository cannot batch or its load fails, the decision path falls back to
	// its existing point queries and remains fail closed.
	scopedRead := constants.IsScopedInternalAccess(ctx, objectName, constants.PermRead)
	readAccessCtx, readAccessCounter := telemetry.WithSQLStatementCounter(ctx)
	readAccessStart := time.Now()
	if !scopedRead {
		ps.permissions.PrimeRecordAccessInputCache(readAccessCtx, schema.APIName, []models.SObject{record}, user)
	}
	readAccessElapsed, readAccessSQL = time.Since(readAccessStart), readAccessCounter.Count()
	readAccessSQLTime, readAccessSQLMax = readAccessCounter.TotalDuration(), readAccessCounter.MaxDuration()

	// 4. Compute one complete access result on the RAW record. Its Read bit is
	// the same authoritative decision that the former standalone Read check made,
	// so use it for 404 existence concealment as well as the returned projection.
	// This avoids evaluating the entire record sharing graph twice.
	allAccessCtx, allAccessCounter := telemetry.WithSQLStatementCounter(ctx)
	allAccessStart := time.Now()
	access := ps.permissions.CheckRecordAccessAll(permissions.WithRecordAccessPhaseProfile(allAccessCtx), schema, record, user)
	allAccessElapsed, allAccessSQL = time.Since(allAccessStart), allAccessCounter.Count()
	allAccessSQLTime, allAccessSQLMax = allAccessCounter.TotalDuration(), allAccessCounter.MaxDuration()
	if !scopedRead && !access.HasReadAccess {
		return nil, nil, apierr.NotFound(fmt.Sprintf("%s with ID '%s' not found", objectName, id))
	}

	// 5. FLS (Field Level Security) — filter fields user cannot see.
	// ElevatedContext+ bypasses FLS (see Retrieve for rationale).
	flsCtx, flsCounter := telemetry.WithSQLStatementCounter(ctx)
	flsStart := time.Now()
	var filtered models.SObject
	if constants.GetSecurityContext(ctx).BypassesCRUDandFLS() ||
		constants.ScopedInternalAccessBypassesFLS(ctx, objectName, constants.PermRead) {
		filtered = ps.filterReadBlockedFields(flsCtx, objectName, record)
	} else {
		filtered = make(models.SObject)
		for k, v := range record {
			readable, err := ps.isReadableUserFieldWithSchema(flsCtx, schema, k, user)
			if err != nil {
				flsElapsed, flsSQL = time.Since(flsStart), flsCounter.Count()
				flsSQLTime, flsSQLMax = flsCounter.TotalDuration(), flsCounter.MaxDuration()
				return nil, nil, err
			}
			if readable {
				filtered[k] = v
			}
		}
		schemautil.CopyExactDecimalsForPresentFields(record, filtered)
	}
	flsElapsed, flsSQL = time.Since(flsStart), flsCounter.Count()
	flsSQLTime, flsSQLMax = flsCounter.TotalDuration(), flsCounter.MaxDuration()

	grantsCtx, grantsCounter := telemetry.WithSQLStatementCounter(ctx)
	grantsStart := time.Now()
	if err := ps.projectPermissionGrants(grantsCtx, objectName, []models.SObject{filtered}); err != nil {
		grantsElapsed, grantsSQL = time.Since(grantsStart), grantsCounter.Count()
		grantsSQLTime, grantsSQLMax = grantsCounter.TotalDuration(), grantsCounter.MaxDuration()
		zlog.Error().Err(err).Str("request_id", reqID).Str("object", objectName).Str("record_id", id).Msg("PersistenceService.RetrieveWithAccess projectPermissionGrants failed")
		return nil, nil, err
	}
	grantsElapsed, grantsSQL = time.Since(grantsStart), grantsCounter.Count()
	grantsSQLTime, grantsSQLMax = grantsCounter.TotalDuration(), grantsCounter.MaxDuration()

	return filtered, &access, nil
}

// RetrieveManyRaw retrieves multiple records by IDs with object-level
// permissions and record-level access (RLS) enforced, but without FLS
// filtering. Records the user cannot access at the row level are silently
// omitted. Callers that need raw field values for sharing-rule evaluation
// (e.g. BatchCheckRecordAccess) should use this instead of GetManyRaw.
func (ps *PersistenceService) RetrieveManyRaw(ctx context.Context, objectName string, ids []string, user *models.UserSession) (map[string]models.SObject, error) {
	if user == nil {
		return nil, apierr.Forbidden("authentication required")
	}
	if len(ids) == 0 {
		return map[string]models.SObject{}, nil
	}

	// Object-level permission check
	scopedRead := constants.IsScopedInternalAccess(ctx, objectName, constants.PermRead)
	privilegedRead := constants.GetSecurityContext(ctx) >= constants.SystemContext || scopedRead
	if ps.permissions != nil {
		if privilegedRead {
			if ps.permissions.IsBlockedObject(ctx, objectName, constants.PermRead) {
				return nil, apierr.Forbidden(fmt.Sprintf("blocked: %s is not accessible via generic %s, even in system context", objectName, constants.PermRead))
			}
		} else {
			if err := ps.permissions.CheckPermissionOrErrorWithUser(ctx, objectName, constants.PermRead, user); err != nil {
				return nil, err
			}
		}
	}

	var schema *models.ObjectMetadata
	if ps.metadata != nil {
		schema = ps.metadata.GetSchema(ctx, objectName)
	}

	// Batch fetch raw records
	rawRecords, err := ps.repo.FindMany(ctx, objectName, ids)
	if err != nil {
		return nil, errors.NewInternalError("failed to retrieve records", err)
	}

	// Filter by record-level access (RLS) only — no FLS filtering.
	result := make(map[string]models.SObject, len(rawRecords))
	if !scopedRead && schema != nil && ps.permissions != nil {
		ps.permissions.PrimeReadAccessInputCache(ctx, schema.APIName, recordsFromMap(rawRecords), user)
	}
	for id, record := range rawRecords {
		if ps.permissions != nil {
			if !scopedRead && schema != nil && !ps.permissions.CheckRecordAccess(ctx, schema, record, constants.PermRead, user) {
				continue // silently omit unauthorized records
			}
		}
		result[id] = record
	}

	records := make([]models.SObject, 0, len(result))
	for _, r := range result {
		records = append(records, r)
	}
	if err := ps.projectPermissionGrants(ctx, objectName, records); err != nil {
		return nil, err
	}

	return result, nil
}

// RetrieveByField retrieves a single record by a specific field with strict permission checks
// SECURITY: Uses parameterized queries internally via repo.FindOneByField, preventing SQL injection
func (ps *PersistenceService) RetrieveByField(ctx context.Context, objectName string, fieldName string, fieldValue string, user *models.UserSession) (models.SObject, error) {
	// 1. Check Object Read Permission
	schema, err := ps.prepareOperation(ctx, objectName, constants.PermRead, user)
	if err != nil {
		return nil, err
	}

	// 2. Fetch Record using parameterized query
	// repo.FindOneByField uses query builder with parameterized WHERE clause
	record, err := ps.repo.FindOneByField(ctx, objectName, fieldName, fieldValue)
	if err != nil {
		return nil, errors.NewInternalError("failed to retrieve record by field", err)
	}
	if record == nil {
		return nil, nil // Not found, return nil without error for handler to return 404
	}

	// 3. Check Record Access (Sharing Rules)
	// Return nil (not 403) to prevent leaking record existence.
	scopedRead := constants.IsScopedInternalAccess(ctx, objectName, constants.PermRead)
	if !scopedRead && !ps.permissions.CheckRecordAccess(ctx, schema, record, constants.PermRead, user) {
		return nil, nil
	}

	// 4. FLS (Field Level Security) — filter fields user cannot see.
	// ElevatedContext+ bypasses FLS (see Retrieve for rationale).
	if constants.GetSecurityContext(ctx).BypassesCRUDandFLS() ||
		constants.ScopedInternalAccessBypassesFLS(ctx, objectName, constants.PermRead) {
		return ps.filterReadBlockedFields(ctx, objectName, record), nil
	}

	filtered := make(models.SObject)
	for k, v := range record {
		readable, err := ps.isReadableUserField(ctx, objectName, k, user)
		if err != nil {
			return nil, err
		}
		if readable {
			filtered[k] = v
		}
	}
	schemautil.CopyExactDecimalsForPresentFields(record, filtered)

	return filtered, nil
}

// RetrieveRecords retrieves multiple records (bulk read)
func (ps *PersistenceService) RetrieveRecords(ctx context.Context, objectName string, ids []string, fields []string, user *models.UserSession) ([]models.SObject, error) {
	if len(ids) == 0 {
		return []models.SObject{}, nil
	}

	// Strict Salesforce Parity: If objectName is empty, we must infer it from IDs.
	// Since we use UUIDs (no key prefix), we must query all tables.
	var idTypeMap map[string]string
	if objectName == "" {
		var err error
		idTypeMap, err = ps.ResolveObjectTypes(ctx, ids)
		if err != nil {
			return nil, err
		}
	}

	results := make([]models.SObject, len(ids))
	for i, id := range ids {
		// Determine which object to retrieve from
		targetObjName := objectName
		if targetObjName == "" {
			if foundType, ok := idTypeMap[id]; ok {
				targetObjName = foundType
			} else {
				// Record not found in any table
				results[i] = nil
				continue
			}
		}

		rec, err := ps.Retrieve(ctx, targetObjName, id, user)
		if err != nil {
			// Salesforce: null for not-found or access-denied
			results[i] = nil
			continue
		}

		models.StampObjectType(rec, targetObjName)

		// Filter fields if requested
		if len(fields) > 0 {
			filtered := make(models.SObject)
			if objectType := models.RecordObjectType(rec); objectType != "" {
				models.StampObjectType(filtered, objectType)
			}
			if v, ok := rec["Id"]; ok {
				filtered["Id"] = v
			}
			for _, f := range fields {
				f = strings.TrimSpace(f)
				if val, ok := rec[f]; ok {
					filtered[f] = val
				}
			}
			schemautil.CopyExactDecimalsForPresentFields(rec, filtered)
			results[i] = filtered
		} else {
			results[i] = rec
		}
	}

	return results, nil
}

// UpsertRecords upserts records by external ID field.
// Salesforce Collections Upsert: PATCH /composite/sobjects/{sObjectName}/{externalIdFieldName}
// For each record: if a record with the given external ID exists, update it; otherwise, create it.
func (ps *PersistenceService) UpsertRecords(ctx context.Context, sObjectName string, externalIdField string, records []models.SObject, allOrNone bool, user *models.UserSession) ([]contract.RecordMutationResult, error) {
	if len(records) == 0 {
		return nil, nil
	}
	if !constants.GetSecurityContext(ctx).BypassesCRUDandFLS() {
		if err := rejectGenericListViewMutation(sObjectName, "collection upsert"); err != nil {
			return nil, err
		}
	}

	// SECURITY: Pre-validate object existence to prevent probing internal tables
	// Upsert is a write operation, so we check for Edit permission implicitly via Update/Insert later,
	// but we must verify it's a valid SObject before any DB read.
	schema := ps.metadata.GetSchema(ctx, sObjectName)
	if schema == nil {
		return nil, errors.NewNotFoundWireError(fmt.Sprintf("Object not found: %s", sObjectName))
	}

	results := make([]contract.RecordMutationResult, len(records))
	objectNames := []string{sObjectName}
	if err := ps.rejectExclusiveSiblingSharedWriteSurface(ctx, fmt.Sprintf("UpsertRecords(allOrNone=%t)", allOrNone), objectNames); err != nil {
		return nil, err
	}

	processFn := func(txCtx context.Context) error {
		for i, rec := range records {
			// Extract external ID value (allow any type)
			extIdValue := rec[externalIdField]
			if extIdValue == nil || (fmt.Sprintf("%v", extIdValue) == "") {
				errMsg := fmt.Sprintf("record %d missing external id field '%s'", i, externalIdField)
				// Populate per-record entry BEFORE returning rollback signal so
				// markAllOrNoneRollback preserves the specific error (otherwise
				// the nil-slot rewrite would mask it as ALL_OR_NONE_*).
				results[i] = processingHaltedEntry("", errMsg)
				if allOrNone {
					return errors.NewValidationError(externalIdField, errMsg)
				}
				continue
			}

			// Prepare data (strip attributes)
			data := make(map[string]interface{})
			for k, v := range rec {
				if !models.IsRecordMetadataKey(k) {
					data[k] = v
				}
			}

			// Try to find existing record by external ID field BEFORE normalization.
			// normalizeFieldNamesForObject may alter field name casing, so we must
			// use the original externalIdField name for the lookup.
			// Provide schema to FindOneByField if possible? No, repo usage pattern is plain.
			// But we validated 'sObjectName' via GetSchema above, so we know it's a valid object.
			existing, err := ps.repo.FindOneByField(txCtx, sObjectName, externalIdField, extIdValue)
			if err != nil && !errors.IsNotFound(err) {
				results[i] = processingHaltedEntry("", err.Error())
				if allOrNone {
					// A DB lookup failure is infra, not a per-record business
					// rollback. Wrap so the outer IsInternal guard catches it
					// and the REST layer emits 500 instead of silently dressing
					// it up as HTTP 200 ALL_OR_NONE_OPERATION_ROLLED_BACK.
					return errors.NewInternalError("upsert lookup failed", err)
				}
				continue
			}

			// Normalize field names AFTER lookup
			data = ps.normalizeFieldNamesForObject(txCtx, sObjectName, data)

			if existing != nil {
				// Update existing record
				var existingId string
				if val, ok := existing["Id"].(string); ok {
					existingId = val
				} else if val, ok := existing["id"].(string); ok {
					existingId = val
				}
				if existingId == "" {
					errMsg := "existing record has no ID"
					results[i] = processingHaltedEntry("", errMsg)
					if allOrNone {
						return errors.NewInternalError(errMsg, fmt.Errorf("upsert: record matched by external ID but has no Id field (object=%s)", sObjectName))
					}
					continue
				}

				// Remove ID fields from update data
				delete(data, "Id")
				delete(data, "id")

				err = ps.Update(txCtx, sObjectName, existingId, data, user)
				if err != nil {
					results[i] = bulkErrorEntry(existingId, err)
					if allOrNone {
						return err
					}
					continue
				}
				results[i] = recordMutationUpsertSuccess(existingId, false)
			} else {
				// Insert new record
				insertResult, err := ps.Insert(txCtx, sObjectName, data, user)
				if err != nil {
					results[i] = bulkErrorEntry("", err)
					if allOrNone {
						return err
					}
					continue
				}

				var idStr string
				if val, ok := insertResult["id"].(string); ok {
					idStr = val
				} else if val, ok := insertResult["Id"].(string); ok {
					idStr = val
				}

				results[i] = recordMutationUpsertSuccess(idStr, true)
			}
		}
		return nil
	}

	if allOrNone {
		err := ps.runGroupMembershipAwarePessimisticTransactionForObjects(ctx, objectNames, processFn)
		if err != nil {
			// Infrastructure failures (tx begin/commit/rollback, IO) must
			// surface — do not mask them as per-record rollbacks.
			if errors.IsInternal(err) {
				return nil, err
			}
			// SF parity: allOrNone=true rollback → HTTP 200 + per-record
			// failure (ALL_OR_NONE_OPERATION_ROLLED_BACK for prior-success).
			// Upsert records may not have resolved IDs yet (matching happens in
			// processFn); fall back to empty id for unprocessed slots.
			markAllOrNoneRollback(results, func(i int) string {
				if id, ok := records[i]["Id"].(string); ok && id != "" {
					return id
				}
				if id, ok := records[i]["id"].(string); ok {
					return id
				}
				return ""
			})
			return results, nil
		}
	} else {
		_ = processFn(ctx)
	}

	return results, nil
}

// projectPermissionGrants hydrates virtual fields on a record set from
// the _System_PermissionSetPermission junction when the object's
// decorated schema declares any SystemPermission virtuals. Metadata-
// driven: no objectName gate, no PermissionSet-specific branch — any
// object whose schema the decorator augments with SystemPermission
// virtuals gets its projection here, and every other object short-
// circuits on the empty VirtualFieldsByKind result. Batches grant
// lookups across all IDs to avoid N+1.
func (ps *PersistenceService) projectPermissionGrants(ctx context.Context, objectName string, records []models.SObject) error {
	// RetrieveMany treats ps.metadata as optional (conditional schema load
	// at line 575); mirror that invariant here. Without the nil check this
	// function panics on ps.metadata.GetSchema when metadata wiring is
	// absent (some test fixtures). Raised by Copilot review.
	if ps.permissions == nil || ps.metadata == nil || len(records) == 0 {
		return nil
	}
	schema := ps.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return nil
	}
	if len(models.VirtualFieldsByKind(schema, constants.VirtualFieldSystemPermission)) == 0 {
		return nil
	}
	ids := make([]string, 0, len(records))
	for _, r := range records {
		if id, ok := r[constants.FieldID].(string); ok && id != "" {
			ids = append(ids, id)
		}
	}
	grantsByID, err := ps.permissions.GetPermissionSetGrantsMany(ctx, ids)
	if err != nil {
		return err
	}
	for _, r := range records {
		id, _ := r[constants.FieldID].(string)
		permissions.ProjectPermissionGrantsOntoRecord(r, schema, grantsByID[id])
	}
	return nil
}