package records

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

	"github.com/nexuscrm/audit"
	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	kernelSchema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// extractRecordID extracts the ID from a record using the SSoT field constant.
// This strictly uses constants.FieldID - no fallback or aliasing.
func extractRecordID(record models.SObject) string {
	if val, ok := record[constants.FieldID]; ok {
		if strVal, ok := val.(string); ok && strVal != "" {
			return strVal
		}
	}
	return ""
}

// ... (existing code skipped)

// UpsertByExternalID creates or updates a record based on an external ID field.
// For unique fields, the operation starts a pessimistic transaction when it has
// no ambient transaction, or participates in the caller's existing transaction.
// If two creators race, only a conflict on the exact selected unique field is
// marked replay-safe: the transaction owner rolls the losing attempt back in
// full, then replays so this method observes and updates the winner. Conflicts
// on the primary key or any other unique field are never adopted.
//
// Non-unique fields retain the legacy read-then-write behavior because there is
// no database identity constraint from which a single winner can be derived.
func (ps *PersistenceService) UpsertByExternalID(
	ctx context.Context,
	objectName string,
	fieldName string,
	fieldValue string,
	data models.SObject,
	currentUser *models.UserSession,
) (models.SObject, bool, error) {
	return ps.upsertByExternalID(ctx, objectName, fieldName, fieldValue, data, false, nil, currentUser)
}

// UpsertByUniqueExternalID is the fail-closed atomic variant used by authored
// Flow metadata. Unlike the legacy-compatible UpsertByExternalID surface, it
// rejects a non-unique field instead of silently falling back to read-then-
// write behavior that cannot choose one winner under concurrency.
func (ps *PersistenceService) UpsertByUniqueExternalID(
	ctx context.Context,
	objectName string,
	fieldName string,
	fieldValue string,
	data models.SObject,
	currentUser *models.UserSession,
) (models.SObject, bool, error) {
	return ps.upsertByExternalID(ctx, objectName, fieldName, fieldValue, data, true, nil, currentUser)
}

// UpsertByExternalIDWithAdoptionGuards performs the same atomic create-or-adopt
// operation as UpsertByExternalID, but permits adoption of an existing unique-
// identity owner only when every named guard is blank or exactly equal to the
// final incoming value. Guard evaluation happens after before-save mutation and
// under the target row's Update lock.
func (ps *PersistenceService) UpsertByExternalIDWithAdoptionGuards(
	ctx context.Context,
	objectName string,
	fieldName string,
	fieldValue string,
	data models.SObject,
	guardFields []string,
	currentUser *models.UserSession,
) (models.SObject, bool, error) {
	return ps.upsertByExternalID(ctx, objectName, fieldName, fieldValue, data, true, guardFields, currentUser)
}

func (ps *PersistenceService) upsertByExternalID(
	ctx context.Context,
	objectName string,
	fieldName string,
	fieldValue string,
	data models.SObject,
	requireUnique bool,
	guardFields []string,
	currentUser *models.UserSession,
) (upserted models.SObject, wasCreated bool, err error) {
	// Rejected-save audit (#4261): upsert is the one surface where suppressing
	// the probe's uniqueness conflict could swallow a real refusal.
	//
	// The attempted INSERT is marked so its uniqueness conflict is not
	// recorded — that conflict is how the upsert discovers the record already
	// exists, and it then adopts it via UPDATE. But an adoption conflict or a
	// lost unique race can ALSO surface a conflict that is terminal: the
	// caller gets an error and nothing landed. Without a hook here that
	// outcome would produce no row at all, because the suppression already
	// silenced the inner attempt and upsert has no other entry point.
	//
	// So the claim is installed at this level and taken here on failure. The
	// inner Insert/Update hooks still record whatever they classify first
	// (a required-field or rule refusal is theirs, not this one's); this only
	// fires when the operation ends in an error nothing else claimed.
	ctx = withRejectedSaveClaim(ctx)
	upsertClaim := rejectedSaveClaimFrom(ctx)
	recordedBefore := upsertClaim.recordedCount()
	upsertRequestedFields := ps.rejectedSaveRequestedFields(data)
	defer func() {
		if err == nil {
			return
		}
		// Last resort only. Identity dedupe is not enough here: the adoption
		// guard re-wraps the failure with a sentinel cause on the way out, so
		// the error this defer sees is NOT errors.Is the one an inner Insert
		// or Update already recorded, and emitting would produce a second row
		// for one refused save. Counting claims across this call's own span
		// is immune to that re-wrapping.
		if upsertClaim.recordedCount() > recordedBefore {
			return
		}
		ps.emitRejectedSaveAudit(ctx, objectName, "", audit.ActionCreate,
			upsertRequestedFields, err, currentUser)
	}()

	// 1. Validate schema exists
	schema := ps.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return nil, false, fmt.Errorf("object %s not found (or schema missing)", objectName)
	}

	// 2. Resolve the canonical external ID field.
	var externalIDField *models.FieldMetadata
	for i := range schema.Fields {
		if strings.EqualFold(schema.Fields[i].APIName, fieldName) {
			externalIDField = &schema.Fields[i]
			break
		}
	}
	if externalIDField == nil {
		return nil, false, fmt.Errorf("Field '%s' does not exist on object '%s'. Please check for typos.", fieldName, objectName)
	}
	canonicalFieldName := externalIDField.APIName
	canonicalGuardFields, canonicalGuardValues, guardErr := canonicalExternalIDUpsertAdoptionGuards(
		schema,
		canonicalFieldName,
		guardFields,
		data,
	)
	if guardErr != nil {
		return nil, false, guardErr
	}
	if (requireUnique || len(canonicalGuardFields) > 0) && !externalIDField.IsUnique {
		return nil, false, fmt.Errorf(
			"guarded external-ID upsert requires unique field %s.%s",
			objectName,
			canonicalFieldName,
		)
	}
	if externalIDField.IsUnique {
		if !externalIDField.Createable || !externalIDField.Updateable || externalIDField.IsVirtual ||
			externalIDField.IsCompound || externalIDField.Formula != nil {
			return nil, false, fmt.Errorf(
				"atomic external-ID upsert requires a createable, updateable, persistable non-formula field %s.%s",
				objectName,
				canonicalFieldName,
			)
		}
	}

	// 3. Build immutable attempt input without mutating the caller's map. Remove
	// case variants so every retry writes one canonical logical field. On the
	// fail-closed unique path, a blank fieldValue means the Flow deliberately
	// deferred identity resolution to the canonical Insert path, where schema
	// defaults and before-save mutation run. The first final insert image binds
	// the identity for all later retries. Legacy-compatible non-unique upsert
	// retains its historical blank-value behavior.
	resolvedFieldValue := fieldValue
	fieldValueResolved := !requireUnique || strings.TrimSpace(fieldValue) != ""
	baseData := make(models.SObject, len(data)+1)
	for k, v := range data {
		if !strings.EqualFold(k, canonicalFieldName) && !containsFold(canonicalGuardFields, k) {
			baseData[k] = v
		}
	}
	if fieldValueResolved {
		baseData[canonicalFieldName] = resolvedFieldValue
	}
	for field, value := range canonicalGuardValues {
		baseData[field] = value
	}

	var result models.SObject
	var created bool
	runUpsert := func(txCtx context.Context) error {
		// Insert/Update mutate their input with system fields. A replay must begin
		// from clean request data after the previous transaction rolled back.
		attemptData := cloneSObject(baseData)
		if fieldValueResolved {
			attemptData[canonicalFieldName] = resolvedFieldValue
		}
		result = nil
		created = false

		// 5. Check if record already exists by external ID
		var existing models.SObject
		if fieldValueResolved {
			var errQuery error
			existing, errQuery = ps.repo.FindOneByField(txCtx, objectName, canonicalFieldName, resolvedFieldValue)
			if errQuery != nil {
				return appErrors.NewInternalError("failed to query by external ID", errQuery)
			}
		}

		// 6. If exists, perform UPDATE
		if existing != nil {
			// Extract ID - check multiple possible key formats due to schema aliasing
			existingID := extractRecordID(existing)
			if existingID == "" {
				return appErrors.NewInternalError("existing record has no ID", fmt.Errorf("upsert: record matched by external ID but has no Id field (object=%s, keys=%v)", objectName, getRecordKeys(existing)))
			}

			// Access check BEFORE calling Update — if the caller lacks access, return
			// 404 to match Salesforce v63 upsert behavior where inaccessible records
			// are indistinguishable from non-existent ones (AC2).
			// Use appErrors.NewNotFoundError because the boundary adapter maps
			// it to the 404 envelope this upsert path needs.
			if constants.GetSecurityContext(txCtx) < constants.SystemContext &&
				!constants.IsScopedInternalAccess(txCtx, objectName, constants.PermEdit) {
				if !ps.permissions.CheckObjectPermissionOnRecord(txCtx, schema, existing, constants.PermEdit, currentUser) {
					return appErrors.NewNotFoundError(objectName, "")
				}
				if !ps.permissions.CheckRecordAccess(txCtx, schema, existing, constants.PermEdit, currentUser) {
					return appErrors.NewNotFoundError(objectName, "")
				}
			}

			// Remove ID from update data if present (should not update ID)
			delete(attemptData, constants.FieldID)
			delete(attemptData, constants.ColID)

			// Perform Update. Propagate the underlying error unchanged so typed
			// business errors (FLSDeniedError, ValidationError, PermissionError)
			// reach the REST layer with their semantic intact. Wrapping in
			// InternalError here would force every handler to first unwrap past
			// the outer infra-like envelope before classifying, and worse,
			// the IsInternal guard in REST handlers would then misreport a
			// legitimate 400 FLS denial as a 500 INTERNAL_ERROR.
			guardIdentityField := ""
			guardIdentityValue := ""
			if externalIDField.IsUnique {
				guardIdentityField = canonicalFieldName
				guardIdentityValue = resolvedFieldValue
			}
			updateCtx, guardObservation := withExternalIDUpsertAdoptionGuardAndIdentity(
				txCtx,
				objectName,
				existingID,
				canonicalGuardFields,
				guardIdentityField,
				guardIdentityValue,
			)
			if err := ps.Update(updateCtx, objectName, existingID, attemptData, currentUser); err != nil {
				if externalIDUpsertAdoptionGuardDirectMutationConflict(err, guardObservation) &&
					exactAdoptionGuardUniqueConflict(err, objectName, schema, canonicalGuardFields) {
					// Expose one semantic sentinel after exact direct-mutation
					// classification. Keep the repository envelope in the message,
					// not the unwrap chain: a successful rollback must return a
					// routable ownership fault, while a rollback failure is wrapped
					// by the transaction manager in a top-level InternalError and
					// therefore remains fail-closed in Flow.
					return newExternalIDUpsertAdoptionConflictError(
						fmt.Sprintf(
							"refusing to adopt %s record %s because the requested guard identity is owned by another record",
							objectName,
							existingID,
						),
						err,
					)
				}
				return err
			}

			attemptData[constants.FieldID] = existingID
			result = attemptData
			created = false
			return nil
		}

		// 7. If not exists, perform INSERT
		var finalAttemptedIdentity any
		var finalAttemptedIdentityObserved bool
		insertCtx, insertObservation := withInsertFinalImageObserver(
			txCtx,
			func(finalImage models.SObject) error {
				finalAttemptedIdentity, finalAttemptedIdentityObserved =
					lookupRecordValueCaseInsensitive(finalImage, canonicalFieldName)
				finalIdentity, ok := finalAttemptedIdentity.(string)
				if requireUnique && (!finalAttemptedIdentityObserved || !ok || strings.TrimSpace(finalIdentity) == "") {
					return fmt.Errorf(
						"atomic external-ID upsert requires a non-blank final string value for %s.%s after defaults and before-save mutation",
						objectName,
						canonicalFieldName,
					)
				}
				if requireUnique && fieldValueResolved && finalIdentity != resolvedFieldValue {
					return fmt.Errorf(
						"atomic external-ID upsert final identity for %s.%s changed from %q to %q",
						objectName,
						canonicalFieldName,
						resolvedFieldValue,
						finalIdentity,
					)
				}
				if requireUnique {
					resolvedFieldValue = finalIdentity
					fieldValueResolved = true
				}
				return nil
			},
		)
		// This INSERT is a probe: the upsert uses a uniqueness conflict as
		// its signal that the record already exists, then adopts it via
		// UPDATE. Mark the attempt so that outcome is not recorded as a
		// rejected save (#4261). Only the uniqueness stage is suppressed —
		// a required-field or rule refusal here fails the whole upsert and
		// is still recorded.
		inserted, err := ps.Insert(withExpectedUniquenessConflict(insertCtx), objectName, attemptData, currentUser)
		if err != nil {
			if externalIDUpsertDirectInsertConflict(err, insertObservation) &&
				exactAdoptionGuardUniqueConflict(err, objectName, schema, canonicalGuardFields) {
				return newExternalIDUpsertAdoptionConflictError(
					fmt.Sprintf(
						"refusing to create or adopt %s because the requested guard identity is owned by another record",
						objectName,
					),
					err,
				)
			}
			if externalIDField.IsUnique && finalAttemptedIdentityObserved &&
				replaySafeUniqueFieldConflict(err, insertObservation, objectName, *externalIDField, finalAttemptedIdentity, resolvedFieldValue) {
				winnerExists, winnerErr := ps.externalIDUpsertLiveWinnerExists(
					txCtx,
					objectName,
					canonicalFieldName,
					resolvedFieldValue,
				)
				if winnerErr != nil {
					return winnerErr
				}
				if winnerExists {
					return fmt.Errorf("upsert %s by %s lost concurrent create race: %w: %w", objectName, canonicalFieldName, ports.ErrUniqueFieldUpsertRace, err)
				}
			}
			if externalIDField.IsUnique && finalAttemptedIdentityObserved &&
				replaySafeLiveWinnerPrecheckConflict(err, insertObservation, objectName, *externalIDField, finalAttemptedIdentity, resolvedFieldValue) {
				winnerExists, winnerErr := ps.externalIDUpsertLiveWinnerExists(
					txCtx,
					objectName,
					canonicalFieldName,
					resolvedFieldValue,
				)
				if winnerErr != nil {
					return winnerErr
				}
				if winnerExists {
					return fmt.Errorf("upsert %s by %s observed concurrent winner during uniqueness precheck: %w: %w", objectName, canonicalFieldName, ports.ErrUniqueFieldUpsertRace, err)
				}
			}
			return err
		}
		result = inserted
		created = true
		return nil
	}
	runWithExistingWriteOrchestration := func(orchestrationCtx context.Context, forceFullBoundary bool) error {
		if schemaHasExclusiveSiblingFields(schema) {
			if boundaryErr := ps.rejectExclusiveSiblingPlainOuterTransaction(orchestrationCtx, "UpsertByExternalID"); boundaryErr != nil {
				return boundaryErr
			}
			return runUpsert(orchestrationCtx)
		}
		if forceFullBoundary {
			// Unique-field upsert owns a whole read/create-or-adopt transaction,
			// but it must still enter the canonical rollup, parent-lock, and
			// group-membership orchestration used by ordinary record writes.
			return ps.runRollupAwareTransaction(orchestrationCtx, objectName, baseData, runUpsert)
		}
		return ps.runRollupAwareTransactionIfNeeded(orchestrationCtx, objectName, baseData, runUpsert)
	}
	if externalIDField.IsUnique {
		if ps.txManager == nil {
			err = appErrors.NewInternalError("unique-field upsert requires a transaction manager", nil)
		} else if ps.txManager.IsTransactionActive(ctx) {
			// The outer transaction remains the only rollback/replay owner. A
			// unique-race marker bubbles to WithRetry/WithPessimisticRetry; a raw
			// transaction owner safely rolls back and may choose whether to retry.
			err = runWithExistingWriteOrchestration(ctx, true)
		} else {
			err = ps.runUniqueFieldUpsertTransaction(ctx, objectName, runUpsert)
		}
	} else {
		err = runWithExistingWriteOrchestration(ctx, false)
	}
	if err != nil {
		return nil, false, err
	}

	return result, created, nil
}

func (ps *PersistenceService) externalIDUpsertLiveWinnerExists(
	ctx context.Context,
	objectName string,
	fieldName string,
	fieldValue string,
) (bool, error) {
	// A normal SELECT inside TiDB REPEATABLE READ reuses the attempt snapshot.
	// After an exact duplicate-key collision that snapshot may still be the one
	// that observed no owner, even though the competing winner has committed.
	// This detached lookup is read-only classification: it does not mutate or
	// commit outside the ambient transaction. The ambient owner still must roll
	// back the losing attempt before WithRetry/WithPessimisticRetry may replay.
	// Querying outside the stale snapshot also preserves the live-row proof that
	// prevents soft-deleted-only unique owners from becoming replay authority.
	liveReadCtx := ctx
	if ps.txManager != nil && ps.txManager.IsTransactionActive(ctx) {
		liveReadCtx = ps.txManager.DetachTx(ctx)
		if ps.txManager.IsTransactionActive(liveReadCtx) {
			return false, appErrors.NewInternalError(
				"failed to detach concurrent external-ID winner read from the losing transaction",
				nil,
			)
		}
	}
	winner, err := ps.repo.FindOneByField(liveReadCtx, objectName, fieldName, fieldValue)
	if err != nil {
		return false, appErrors.NewInternalError("failed to verify concurrent external-ID winner", err)
	}
	return winner != nil && extractRecordID(winner) != "", nil
}

func canonicalExternalIDUpsertAdoptionGuards(
	schema *models.ObjectMetadata,
	externalIDField string,
	guardFields []string,
	data models.SObject,
) ([]string, map[string]any, error) {
	if len(guardFields) == 0 {
		return nil, nil, nil
	}

	canonicalFields := make([]string, 0, len(guardFields))
	values := make(map[string]any, len(guardFields))
	for _, authoredField := range guardFields {
		authoredField = strings.TrimSpace(authoredField)
		if authoredField == "" {
			return nil, nil, fmt.Errorf("guarded external-ID upsert contains a blank adoption guard field")
		}
		var guardField *models.FieldMetadata
		for i := range schema.Fields {
			if strings.EqualFold(schema.Fields[i].APIName, authoredField) {
				guardField = &schema.Fields[i]
				break
			}
		}
		if guardField == nil {
			return nil, nil, fmt.Errorf("Field '%s' does not exist on object '%s'. Please check for typos.", authoredField, schema.APIName)
		}
		canonicalField := guardField.APIName
		if strings.EqualFold(canonicalField, externalIDField) {
			return nil, nil, fmt.Errorf("external-ID field %s.%s cannot also be an adoption guard", schema.APIName, canonicalField)
		}
		if containsFold(canonicalFields, canonicalField) {
			return nil, nil, fmt.Errorf("duplicate adoption guard field %s.%s", schema.APIName, canonicalField)
		}
		if !guardField.IsUnique {
			return nil, nil, fmt.Errorf("adoption guard field %s.%s must be unique", schema.APIName, canonicalField)
		}
		if strings.EqualFold(canonicalField, constants.FieldID) || guardField.IsSystem ||
			!guardField.Createable || !guardField.Updateable || guardField.IsVirtual ||
			guardField.IsCompound || guardField.Formula != nil {
			return nil, nil, fmt.Errorf("adoption guard field %s.%s must be a createable, mutable, persistable non-system field", schema.APIName, canonicalField)
		}

		var value any
		matches := 0
		for dataField, dataValue := range data {
			if strings.EqualFold(strings.TrimSpace(dataField), canonicalField) {
				matches++
				value = dataValue
			}
		}
		if matches != 1 || isBlankExternalIDUpsertAdoptionGuardValue(value) {
			return nil, nil, fmt.Errorf("guarded external-ID upsert requires exactly one non-blank value for %s.%s", schema.APIName, canonicalField)
		}
		canonicalFields = append(canonicalFields, canonicalField)
		values[canonicalField] = value
	}
	return canonicalFields, values, nil
}

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

// runUniqueFieldUpsertTransaction owns the whole read/create-or-adopt retry
// boundary for a root unique-field upsert. It deliberately enters the
// source-aware group-membership path before opening the pessimistic
// transaction, then installs the canonical parent-lock coordinator. Insert or
// Update retain their ordinary nested write orchestration inside this boundary.
func (ps *PersistenceService) runUniqueFieldUpsertTransaction(
	ctx context.Context,
	objectName string,
	fn func(context.Context) error,
) error {
	return ps.runGroupMembershipAwarePessimisticTransaction(ctx, objectName, func(txCtx context.Context) error {
		return ps.runParentLockCoordinatorIfActive(txCtx, fn)
	})
}

func replaySafeUniqueFieldConflict(
	err error,
	insertObservation *insertFinalImageObservation,
	objectName string,
	field models.FieldMetadata,
	attemptedValue any,
	requestedValue string,
) bool {
	attemptedString, ok := attemptedValue.(string)
	return externalIDUpsertDirectInsertErrorHasPhase(err, insertObservation, externalIDUpsertInsertRepositoryWrite) &&
		ok &&
		attemptedString == requestedValue &&
		ports.IsDatabaseDuplicateKey(err) &&
		exactUniqueFieldConflict(err, objectName, field)
}

func replaySafeLiveWinnerPrecheckConflict(
	err error,
	insertObservation *insertFinalImageObservation,
	objectName string,
	field models.FieldMetadata,
	attemptedValue any,
	requestedValue string,
) bool {
	attemptedString, ok := attemptedValue.(string)
	return externalIDUpsertDirectInsertErrorHasPhase(err, insertObservation, externalIDUpsertInsertUniquenessPrecheck) &&
		ok &&
		attemptedString == requestedValue &&
		!ports.IsDatabaseDuplicateKey(err) &&
		exactUniqueFieldConflict(err, objectName, field)
}

func externalIDUpsertDirectInsertConflict(err error, insertObservation *insertFinalImageObservation) bool {
	return externalIDUpsertDirectInsertErrorHasPhase(err, insertObservation, externalIDUpsertInsertUniquenessPrecheck) ||
		externalIDUpsertDirectInsertErrorHasPhase(err, insertObservation, externalIDUpsertInsertRepositoryWrite)
}

func externalIDUpsertDirectInsertErrorHasPhase(
	err error,
	insertObservation *insertFinalImageObservation,
	phase externalIDUpsertInsertErrorPhase,
) bool {
	if insertObservation == nil {
		return false
	}
	var directErr *externalIDUpsertDirectInsertError
	return errors.As(err, &directErr) && directErr != nil &&
		directErr.owner == insertObservation && directErr.phase == phase
}

func exactUniqueFieldConflict(err error, objectName string, field models.FieldMetadata) bool {
	var conflict *appErrors.ConflictError
	if !errors.As(err, &conflict) || conflict == nil {
		return false
	}
	if conflict.Resource != "" && !strings.EqualFold(conflict.Resource, objectName) {
		return false
	}

	physicalName := strings.TrimSpace(field.PhysicalName)
	if physicalName == "" {
		physicalName = field.APIName
	}
	candidates := []string{
		field.APIName,
		physicalName,
		kernelSchema.GenerateUniqueIndexName(objectName, field.APIName),
		kernelSchema.GenerateUniqueIndexName(objectName, physicalName),
	}
	for _, candidate := range candidates {
		if strings.EqualFold(strings.TrimSpace(conflict.Field), candidate) {
			return true
		}
	}
	return false
}

func exactAdoptionGuardUniqueConflict(
	err error,
	objectName string,
	schema *models.ObjectMetadata,
	guardFields []string,
) bool {
	if schema == nil || len(guardFields) == 0 {
		return false
	}
	for _, guardField := range guardFields {
		for i := range schema.Fields {
			field := schema.Fields[i]
			if strings.EqualFold(field.APIName, guardField) && field.IsUnique &&
				exactUniqueFieldConflict(err, objectName, field) {
				return true
			}
		}
	}
	return false
}

// getRecordKeys returns the keys of a record for debugging
func getRecordKeys(record models.SObject) []string {
	keys := make([]string, 0, len(record))
	for k := range record {
		keys = append(keys, k)
	}
	return keys
}