package records

import (
	"context"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
)

// GetRecordType resolves the persisted object API name for a record ID.
func (ps *PersistenceService) GetRecordType(ctx context.Context, id string) (string, error) {
	return ps.repo.GetRecordType(ctx, id)
}

// GetRaw retrieves a record by ID without any permission checks.
// SECURITY: Only for system-internal use (bootstrap, event handlers, background jobs).
// User-facing code MUST use Retrieve() which enforces object perms + record access + FLS.
func (ps *PersistenceService) GetRaw(ctx context.Context, objectName string, id string) (models.SObject, error) {
	return ps.repo.FindOne(ctx, objectName, id)
}

func (ps *PersistenceService) cachedFindOne(ctx context.Context, tableName, id string) (models.SObject, error) {
	return ps.cachedFindOneWithMode(ctx, tableName, id, recordReadModeLogical)
}

func (ps *PersistenceService) cachedPhysicalFindOne(ctx context.Context, tableName, id string) (models.SObject, error) {
	return ps.cachedFindOneWithMode(ctx, tableName, id, recordReadModePhysical)
}

func (ps *PersistenceService) cachedFindOneWithMode(ctx context.Context, tableName, id string, mode recordReadMode) (models.SObject, error) {
	c := recordReadCacheFromCtx(ctx)
	readCtx := ctx
	if mode == recordReadModePhysical {
		readCtx = persistence.WithPhysicalReadOnly(ctx)
	}
	if c == nil {
		return ps.repo.FindOne(readCtx, tableName, id)
	}
	if record, found, hit := c.getWithMode(tableName, id, mode); hit {
		if !found {
			return nil, nil
		}
		return record, nil
	}
	readGen := c.beginRead(tableName, id)
	record, err := ps.repo.FindOne(readCtx, tableName, id)
	if err != nil {
		return nil, err
	}
	c.setIfFreshWithMode(tableName, id, mode, record, record != nil, readGen)
	return record, nil
}

// GetLock retrieves a record by ID with a row lock (SELECT ... FOR UPDATE).
// Must be called inside a transaction. This is a checked escape hatch for
// inventoried scheduler/system/target mutation callers; business parent-row
// locks must use records admission paths instead of adding new public callers.
func (ps *PersistenceService) GetLock(ctx context.Context, tableName string, id string) (models.SObject, error) {
	return ps.repo.GetLock(ctx, tableName, id)
}

// DetachTx returns a derived context with the active database transaction
// removed. The returned context retains all other values (security context,
// trace IDs, re-entry guards, etc.) but causes RunInTransaction to start a
// fresh, independent transaction instead of reusing the caller's transaction.
//
// Use this when an automation step must issue a DB write on the SAME row that
// the enclosing transaction already holds a lock on — for example, an
// AfterSave flow that calls recordUpdate on the triggering record while
// running inside the outbox worker's transaction. Calling persistence.Update
// without detaching first would cause a TiDB lock-wait timeout (Error 1205)
// because the SELECT ... FOR UPDATE in loadAndValidateAccess would block
// waiting for a lock already held by the enclosing transaction on the same
// connection.
func (ps *PersistenceService) DetachTx(ctx context.Context) context.Context {
	return ps.txManager.DetachTx(ctx)
}

// PhysicalDeleteRaw permanently removes a record without permissions, soft-delete,
// recycle-bin, cascade, or event handling. For generic runtime-owned record
// shapes, it still checks for existing live children first so raw delete does not
// bypass the historical-reference guard that the normal delete lifecycle runs.
// SECURITY: Only for system-internal replacement flows that own their child rows.
func (ps *PersistenceService) PhysicalDeleteRaw(ctx context.Context, objectName string, id string) error {
	if groupMembershipClosureSourceObject(objectName) {
		return appErrors.NewInternalError("raw physical delete would bypass group membership closure maintenance", nil)
	}
	managedParentField, managedRoot, managedDefinitionProtected, err :=
		managedPermissionSetDefinitionPhysicalDeleteContract(objectName)
	if err != nil {
		return appErrors.NewInternalError("raw physical delete cannot load managed Permission Set definition policy", err)
	}
	var schema *models.ObjectMetadata
	if ps.metadata != nil {
		schema = ps.metadata.GetSchema(ctx, objectName)
	}
	var beforePurgeCapabilities []string
	if schema != nil {
		beforePurgeCapabilities = schema.RecordWriteValidationCapabilitiesForPhase(string(recordWriteValidationBeforePurge))
	}
	// The managed definition is the required side of the security metadata
	// reconciliation contract. Enter the protected path even when the effective
	// schema is stale and has dropped every before_purge capability, so the
	// shared validator runner can report the missing required capability before
	// raw DML instead of silently bypassing validation.
	writeValidationProtected := len(beforePurgeCapabilities) > 0 ||
		len(securityWriteValidationMissingRequiredCapabilities(
			objectName,
			recordWriteValidationBeforePurge,
			beforePurgeCapabilities,
		)) > 0
	permissionStateProtected := ps.permissionStateDeleteRequiresPessimisticTransaction(objectName) ||
		managedDefinitionProtected || writeValidationProtected
	if liveParentGuardSchemaHasCoveredWriter(schema) || permissionStateProtected {
		if ps.txManager == nil {
			return appErrors.NewInternalError("raw physical delete requires transaction manager", nil)
		}
		return ps.runGroupMembershipAwarePessimisticTransaction(ctx, objectName, func(txCtx context.Context) error {
			if liveParentGuardSchemaHasCoveredWriter(schema) {
				if _, err := ps.LockParentRows(txCtx, []recordlocks.ParentKey{{ObjectName: objectName, RecordID: id}}); err != nil {
					return err
				}
				if err := ps.checkAnyLiveChildReference(txCtx, schema, objectName, id); err != nil {
					return err
				}
			}
			if writeValidationProtected {
				validationPreflight, err := ps.repo.FindAny(
					persistence.WithPhysicalReadOnly(txCtx), objectName, id,
				)
				if err != nil {
					return err
				}
				if validationPreflight != nil {
					if err := ps.prelockDeleteParentRows(txCtx, objectName, id, validationPreflight); err != nil {
						return err
					}
				}
			}
			var managedPreflight models.SObject
			if managedDefinitionProtected {
				managedPreflight, err = ps.repo.FindAny(
					persistence.WithPhysicalReadOnly(txCtx),
					objectName,
					id,
				)
				if err != nil {
					return err
				}
				if managedPreflight != nil {
					if !managedRoot {
						parentKeys, err := ps.managedPermissionSetDefinitionParentKeys(
							txCtx,
							objectName,
							contract.Record(managedPreflight),
						)
						if err != nil {
							return err
						}
						if _, err := ps.LockParentRows(txCtx, parentKeys); err != nil {
							return err
						}
					}
					if err := ps.validateManagedPermissionSetDefinitionWriteContract(txCtx, contract.RecordWriteValidationEvent{
						ObjectName: objectName,
						RecordID:   id,
						OldRecord:  contract.Record(managedPreflight),
					}); err != nil {
						return err
					}
				}
			}
			var before models.SObject
			if permissionStateProtected {
				before, err = ps.repo.GetLock(txCtx, objectName, id)
				if err != nil {
					return err
				}
				if before == nil {
					return nil
				}
			}
			if writeValidationProtected && before != nil {
				if err := ps.runRecordWriteValidators(txCtx, recordWriteValidationBeforePurge, recordWriteValidationEvent{
					ObjectName: objectName,
					RecordID:   id,
					Record:     before,
					OldRecord:  before,
				}); err != nil {
					return err
				}
			}
			if managedDefinitionProtected {
				if managedRoot {
					if err := ps.validateManagedPermissionSetDefinitionWriteContract(txCtx, contract.RecordWriteValidationEvent{
						ObjectName: objectName,
						RecordID:   id,
						OldRecord:  contract.Record(before),
					}); err != nil {
						return err
					}
				} else {
					preflightParent := permissionSetDefinitionStringField(managedPreflight, managedParentField)
					lockedParent := permissionSetDefinitionStringField(before, managedParentField)
					if preflightParent != lockedParent {
						if lockedParent == "" {
							return appErrors.NewInternalError("managed Permission Set definition parent changed during raw physical delete", nil)
						}
						return &parentLockPrelockRetryError{
							keys: []recordlocks.ParentKey{{
								ObjectName: constants.TablePermissionSet,
								RecordID:   lockedParent,
							}},
							reason: "managed Permission Set definition parent",
						}
					}
				}
			}
			if err := ps.repo.PhysicalDelete(txCtx, objectName, id); err != nil {
				return err
			}
			if err := ps.requestPermissionStateAdvanceForMutation(txCtx, objectName, before, nil); err != nil {
				return err
			}
			InvalidateRecordReadCache(txCtx, objectName, id)
			return nil
		})
	}
	if err := ps.repo.PhysicalDelete(ctx, objectName, id); err != nil {
		return err
	}
	InvalidateRecordReadCache(ctx, objectName, id)
	return nil
}

// GetManyRaw retrieves multiple records by IDs without any permission checks.
// SECURITY: Only for system-internal use. User-facing code MUST use RetrieveMany().
func (ps *PersistenceService) GetManyRaw(ctx context.Context, objectName string, ids []string) (map[string]models.SObject, error) {
	return ps.repo.FindMany(ctx, objectName, ids)
}

// LockDerivedRecord acquires a parent-row lock for a derived write before the
// rollup engine calculates the new aggregate value. The caller must already
// supply a transaction-bearing ctx; this method only widens security context so
// the platform-owned lock read bypasses user-facing access gates.
func (ps *PersistenceService) LockDerivedRecord(ctx context.Context, objectName, recordID string) error {
	return ps.lockDerivedRecordParent(ctx, objectName, recordID)
}

func (ps *PersistenceService) GetDerivedRecordForUpdate(ctx context.Context, objectName, recordID string) (models.SObject, error) {
	if err := ps.lockDerivedRecordParent(ctx, objectName, recordID); err != nil {
		return nil, err
	}
	readCtx := persistence.WithPhysicalReadOnly(constants.WithSecurityContext(ctx, constants.SystemContext, "derived_field_recompute_read"))
	row, err := ps.repo.FindOne(readCtx, objectName, recordID)
	if err != nil {
		return nil, appErrors.NewInternalError("failed to read derived record", err)
	}
	if row == nil {
		return nil, appErrors.NewNotFoundError("Record", "")
	}
	return row, nil
}

func (ps *PersistenceService) lockDerivedRecordParent(ctx context.Context, objectName, recordID string) error {
	lock := func(lockCtx context.Context) error {
		_, err := ps.LockParentRows(lockCtx, []recordlocks.ParentKey{{ObjectName: objectName, RecordID: recordID}})
		return err
	}
	if recordlocks.ScopeFromContext(ctx) != nil {
		return lock(ctx)
	}
	return ps.RunWithParentLockCoordinator(ctx, lock)
}