package permissions

import (
	"context"
	"database/sql"
	"fmt"
	"strings"

	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/rs/zerolog/log"
)

// --- Record-Level Access Checks ---

// CheckRecordAccess checks if a user can access a specific record
// This checks record-level sharing rules and ownership
func (ps *PermissionService) CheckRecordAccess(ctx context.Context, schema *models.ObjectMetadata, record models.SObject, operation string, user *models.UserSession) bool {
	return ps.checkRecordAccessInternal(ctx, schema, record, operation, user, nil, nil)
}

// checkRecordAccessInternal is the recursive implementation of CheckRecordAccess.
// The visited map tracks object names to prevent infinite recursion in ControlledByParent chains.
// approvalLocked, when non-nil, is a pre-fetched result from isApprovalLockedForOperation;
// the internal probe is skipped. nil means "unknown — probe now". Public callers (CheckRecordAccess)
// always pass nil; batch/aggregate callers pass the pre-fetched value to eliminate the double-probe.
func (ps *PermissionService) checkRecordAccessInternal(ctx context.Context, schema *models.ObjectMetadata, record models.SObject, operation string, user *models.UserSession, visited map[string]bool, approvalLocked *bool) bool {
	if user == nil {
		return false
	}

	// Use the pre-fetched approval-lock value if provided; otherwise probe now.
	// This eliminates the double-probe race when CheckRecordAccessAll / BatchCheckRecordAccess
	// have already called isApprovalLockedForOperation for the same (record, op, user).
	// isApprovalLockedForOperation already honors canBypassApprovalLock internally, so
	// SuperUser / system-admin callers see lockedByApproval=false and pass through.
	var lockedByApproval bool
	if approvalLocked != nil {
		lockedByApproval = *approvalLocked
	} else {
		lockedByApproval = ps.isApprovalLockedForOperation(ctx, record, operation, user)
	}
	if lockedByApproval {
		return false
	}

	// LockRule enforcement: for mutation operations in user-context, if the
	// admin-authored LockRule formula evaluates true the record is read-only.
	// Only trusted ServiceContext+ invocations bypass this. System administrators
	// and SuperUsers are intentionally NOT bypassed — LockRule is an
	// admin-authored metadata invariant and bypassing it for admins would gut
	// the declaration. See canBypassLockRule rationale.
	//
	// This gate intentionally runs BEFORE the SuperUser / ModifyAllData bypasses
	// below so that admins/SuperUsers are also blocked by an active LockRule.
	//
	// This is the authoritative write-side gate: we call evaluateLockRuleForGate
	// so that any formula-engine error causes a fail-closed deny rather than a
	// silent allow. evaluateLockRule (fail-open) is used only for the UI
	// IsLocked signal path where a backend write check still guards the mutation.
	if operation != constants.PermRead && schema != nil && schema.LockRule != nil &&
		!canBypassLockRule(ctx, user) {
		locked, err := ps.evaluateLockRuleForGate(schema, record, user)
		if err != nil || locked {
			return false
		}
	}

	// Layer 2a: System-level ViewAllData / ModifyAllData (global, all objects)
	// ModifyAllData is also the neutral super-user capability. It is checked
	// after LockRule so admin-authored invariants still hold for capability
	// holders.
	if ps.CheckSystemPermission(ctx, user, string(constants.PermModifyAllData)) {
		return true
	}
	if ps.CheckSystemPermission(ctx, user, string(constants.PermViewAllData)) && operation == constants.PermRead {
		return true
	}

	// L0-D D1: classify the object's row visibility once via the app-blind
	// translation point. RowVisibilityFor is nil-safe (nil ⇒ Private), and every
	// downstream branch keeps its own `schema != nil` guard so the nil-schema
	// behavior is byte-identical.
	rowVis := models.RowVisibilityFor(schema)

	// Layer 2b: Object-level ViewAllRecords / ModifyAllRecords (per-object).
	// Parent-controlled children inherit record reach through the parent edge;
	// stale self OP rows must not short-circuit that single read path.
	if schema != nil && rowVis.Kind != models.RowVisibilityParentControlled && !schema.UsesDerivedAccessPolicy() {
		objPerm, err := ps.loadEffectiveObjectPermission(ctx, user, schema.APIName)
		if err == nil && objPerm != nil {
			if objPerm.PermissionsModifyAllRecords {
				return true
			}
			if objPerm.PermissionsViewAllRecords && operation == constants.PermRead {
				return true
			}
		}
	}

	// Extract record ID early (needed by both FolderSharing and standard sharing checks)
	recordID := ""
	if id, ok := record[constants.FieldID]; ok {
		if idStr, ok := id.(string); ok {
			recordID = idStr
		}
	}

	if operation == constants.PermRead && schema != nil {
		if policy := schema.RowReadPolicy(); policy != nil && strings.EqualFold(strings.TrimSpace(policy.Kind), "targetRecordRead") {
			if ps.checkTargetRecordReadAccess(ctx, schema, record, user, visited) {
				return true
			}
			if !policy.IncludeSourceVisibility {
				return false
			}
		} else if isOwnerVisibilityRowReadPolicy(policy) {
			return checkOwnerVisibilityReadAccess(policy, record, user)
		}
	}

	// OWD (Organization-Wide Default) check — must match BuildRLSResult logic.
	// L0-D D1: dispatch on the app-blind row-visibility classification. PublicRead
	// and PublicReadWrite stay DISTINCT (write is only granted by ReadWrite).
	if schema != nil {
		if rowVis.Kind == models.RowVisibilityPublicReadWrite {
			return true
		}
		if rowVis.Kind == models.RowVisibilityPublicRead && operation == constants.PermRead {
			return true
		}
		// FolderScoped: delegate to FolderAccessService — standard record sharing
		// doesn't apply. FolderScopeField == "" (FolderSharing) uses the record's
		// own Id as the folder subject; FolderScopeField == "FolderId"
		// (FolderContentSharing) derives access from the record's FolderId column.
		// Read requires folder View; Edit/Delete requires folder Manage.
		if rowVis.Kind == models.RowVisibilityFolderScoped {
			if ps.folderAccess == nil {
				return false // fail-closed
			}
			subject := recordID
			if rowVis.FolderScopeField != "" {
				// FolderContentSharing: derive the folder from the scope column; an
				// empty reference denies — byte-identical to the original
				// FolderContentSharing empty-FolderId guard. FolderSharing
				// (FolderScopeField == "") uses recordID directly with NO empty
				// guard, matching the original FolderSharing arm exactly.
				subject, _ = record[rowVis.FolderScopeField].(string)
				if subject == "" {
					return false
				}
			}
			required := FolderAccessView
			if operation != constants.PermRead {
				required = FolderAccessManage
			}
			return ps.folderAccess.CheckAccess(ctx, user, subject, required)
		}
		// ControlledByParent: access derived from the parent record.
		// If user has access to the parent, they have access to the child.
		// Falls through to ownership/sharing checks if parent access fails (owner fallback).
		//
		if rowVis.Kind == models.RowVisibilityParentControlled {
			parentAccess := ps.checkControlledByParentAccessDetailed(ctx, schema, record, operation, user, visited)
			if parentAccess.allowed {
				return true
			}
			if parentAccess.malformed || parentAccess.parentOnly {
				return false
			}
			// Optional/standalone parent-controlled records retain owner/share fallback.
		}
	}

	// Extract owner ID from record for subsequent checks
	var ownerIDStr string
	ownerID, hasOwner := record[constants.FieldOwnerID]
	if hasOwner {
		// handle both string and pointer types just in case
		switch v := ownerID.(type) {
		case string:
			ownerIDStr = v
		case *string:
			if v != nil {
				ownerIDStr = *v
			}
		default:
			ownerIDStr = fmt.Sprintf("%v", v)
		}
	}

	// 1. Ownership Check (User or Group)
	if hasOwner {
		// A. User Ownership
		if ownerIDStr == user.ID {
			return true
		}

		// B. Group Ownership (Queue)
		// Check if the ownerID is a Group that the user is a member of.
		// Routed through request-scoped point memoization (issue #1585), so
		// repeated (group, user) checks are reused without preloading every
		// group the user belongs to.
		if ps.isUserInGroupForSession(ctx, ownerIDStr, user.ID, user.ID) {
			return true
		}

	}

	// 2. Role Hierarchy Check (Read-Only access for managers)
	// If the object has EnableHierarchySharing set and the user's role is above the
	// record owner's role in the hierarchy, grant READ access (not edit/delete)
	if operation == constants.PermRead && hasOwner && ownerIDStr != "" && rowVis.HierarchyGranted {
		ownerRoleID := ps.getRecordOwnerRoleID(ctx, ownerIDStr)
		isAbove := ps.isUserAboveInHierarchy(user.RoleID, ownerRoleID)
		if isAbove {
			return true
		}
	}

	// 3. Sharing Rules Check
	// Evaluate all sharing rules for this object
	if schema != nil {
		rules := ps.getCachedSharingRules(ctx, schema.APIName)
		for _, rule := range rules {
			if ps.checkSharingRuleAccess(ctx, record, rule, user, operation) {
				return true
			}
		}
	}

	// 4. Manual Record Share Check (_System_RecordShare)
	if schema != nil && recordID != "" {
		if ps.checkManualShareAccess(ctx, schema.APIName, recordID, user, operation) {
			return true
		}
	}

	// 5. Team Member Check (_System_TeamMember)
	if schema != nil && recordID != "" {
		if ps.checkTeamMemberAccess(ctx, schema.APIName, recordID, user, operation) {
			return true
		}
	}

	// Default to DENY if not owner, not above in hierarchy, and no sharing rule matches
	return false
}

func (ps *PermissionService) checkTargetRecordReadAccess(ctx context.Context, schema *models.ObjectMetadata, record models.SObject, user *models.UserSession, visited map[string]bool) bool {
	if schema == nil {
		return false
	}
	policy := schema.RowReadPolicy()
	if policy == nil || !strings.EqualFold(strings.TrimSpace(policy.Kind), "targetRecordRead") {
		return false
	}
	if ps.metadata == nil || ps.repo == nil || record == nil {
		return false
	}
	if visited == nil {
		visited = make(map[string]bool)
	}
	visitKey := strings.ToLower(strings.TrimSpace(schema.APIName))
	if visitKey != "" {
		if visited[visitKey] {
			return false
		}
		visited[visitKey] = true
		defer delete(visited, visitKey)
	}

	associations, valid := policy.TargetRecordReadAssociations()
	if !valid {
		return false
	}
	associationPresent := false
	for _, association := range associations {
		targetObjectField := association.TargetObjectField
		targetRecordField := association.TargetRecordField
		targetRecordID := recordStringValue(record[targetRecordField])
		if targetRecordID == "" {
			continue
		}
		associationPresent = true
		targetObject := recordStringValue(record[targetObjectField])
		if targetObject == "" || strings.EqualFold(targetObject, schema.APIName) {
			return false
		}
		if !targetRecordReadAllowsTarget(schema, targetRecordField, targetObject) {
			return false
		}
		targetSchema := ps.metadata.GetSchema(ctx, targetObject)
		if !ps.canReadTargetRecordReadSchema(ctx, schema, targetSchema, user) {
			return false
		}
		targetRecord, err := ps.repo.GetRecordByID(ctx, targetObject, targetRecordID)
		if err != nil || targetRecord == nil {
			return false
		}
		if !ps.checkRecordAccessInternal(ctx, targetSchema, targetRecord, constants.PermRead, user, visited, nil) {
			return false
		}
	}
	return associationPresent
}

func recordStringValue(value interface{}) string {
	switch v := value.(type) {
	case nil:
		return ""
	case string:
		return strings.TrimSpace(v)
	case *string:
		if v == nil {
			return ""
		}
		return strings.TrimSpace(*v)
	default:
		return strings.TrimSpace(fmt.Sprintf("%v", v))
	}
}

// checkManualShareAccess checks if user has access via manual record share
func (ps *PermissionService) checkManualShareAccess(ctx context.Context, objectAPIName, recordID string, user *models.UserSession, operation string) bool {
	// Fast path: consult exact request-scoped levels preloaded for the current
	// candidate record set. If this record was not preloaded, fall through to
	// the same level-aware point lookup as before.
	if allowed, ok := ps.cachedManualShareAccess(ctx, objectAPIName, user.ID, recordID, operation); ok {
		return allowed
	}

	// A complete Read/Edit/Delete decision asks this same primitive up to three
	// times. Reuse only within that aggregate decision; a later decision must
	// observe a fresh share snapshot.
	load := func() ([]string, error) {
		return ps.repo.GetManualShareAccessLevels(ctx, objectAPIName, recordID, user.ID)
	}
	var levels []string
	var err error
	if cache := recordAccessEvaluationCacheFromContext(ctx); cache != nil {
		levels, err = cache.manualShareAccessLevels(ctx, objectAPIName, recordID, user.ID, load)
	} else {
		levels, err = load()
	}
	if err != nil {
		return false
	}

	for _, level := range levels {
		if ps.accessLevelAllowsOperation(level, operation) {
			return true
		}
	}
	return false
}

// checkTeamMemberAccess checks if user is a team member with access
func (ps *PermissionService) checkTeamMemberAccess(ctx context.Context, objectAPIName, recordID string, user *models.UserSession, operation string) bool {
	// Fast path: exact request-scoped level, shared by Read/Edit/Delete.
	if allowed, ok := ps.cachedTeamMemberAccess(ctx, objectAPIName, user.ID, recordID, operation); ok {
		return allowed
	}

	load := func() (*string, error) {
		return ps.repo.GetTeamMemberAccessLevel(ctx, objectAPIName, recordID, user.ID)
	}
	var accessLevel *string
	var err error
	if cache := recordAccessEvaluationCacheFromContext(ctx); cache != nil {
		accessLevel, err = cache.teamMemberAccessLevel(ctx, objectAPIName, recordID, user.ID, load)
	} else {
		accessLevel, err = load()
	}
	if err != nil {
		return false
	}
	if accessLevel == nil {
		return false
	}
	return ps.accessLevelAllowsOperation(*accessLevel, operation)
}

// accessLevelAllowsOperation checks if access level permits the operation
func (ps *PermissionService) accessLevelAllowsOperation(accessLevel, operation string) bool {
	// Salesforce-native format only
	switch accessLevel {
	case "All", "Full":
		return true
	case constants.PermEdit:
		return operation == constants.PermRead || operation == constants.PermEdit
	case constants.PermRead:
		return operation == constants.PermRead
	default:
		return false
	}
}

// --- Aggregate Record Access (Read + Edit + Delete) ---

// RecordAccessResult holds read/edit/delete access flags for a single record.
// IsLocked is a generic, application-neutral read-only signal that combines
// the platform's approval-pending lock with any admin-authored
// LockRule formula declared on ObjectMetadata. Frontend treats it as
// the single signal to suppress mutation UI on the record.
type RecordAccessResult struct {
	HasReadAccess   bool
	HasEditAccess   bool
	HasDeleteAccess bool
	IsLocked        bool
}

// CheckRecordAccessAll evaluates Read, Edit, and Delete access for a single record.
// It combines object-level CRUD permissions with record-level sharing checks.
// This is the single-record path; for list views, use BatchCheckRecordAccess.
func (ps *PermissionService) CheckRecordAccessAll(
	ctx context.Context,
	schema *models.ObjectMetadata,
	record models.SObject,
	user *models.UserSession,
) RecordAccessResult {
	ctx = withRecordAccessEvaluationCache(ctx)
	ctx = WithAccessInputCache(ctx)
	profile := newRecordAccessPhaseProfile(ctx)
	defer profile.emit(schema, user)
	// Aggregate access evaluates the same sharing-rule identities for Read,
	// Edit and Delete. Once two or more distinct group targets are present, one
	// request-scoped membership projection is strictly fewer round trips than
	// the historical per-group probes. Single-operation reads retain the point
	// path; this optimization is scoped to callers that explicitly requested the
	// complete access decision.
	profile.bool(ctx, "sharing_groups", func(stageCtx context.Context) bool {
		ps.primeAggregateSharingGroupMembership(stageCtx, schema, record, user)
		return true
	})

	// Object-level permission gates: even if record-level grants access,
	// the user must also have object-level CRUD permission.
	objectName := ""
	if schema != nil {
		objectName = schema.APIName
	}
	hasObjEdit := profile.bool(ctx, "object_edit", func(stageCtx context.Context) bool {
		return objectName != "" && ps.CheckObjectPermissionOnRecord(stageCtx, schema, record, constants.PermEdit, user)
	})
	hasObjDelete := profile.bool(ctx, "object_delete", func(stageCtx context.Context) bool {
		return objectName != "" && ps.CheckObjectPermissionOnRecord(stageCtx, schema, record, constants.PermDelete, user)
	})

	// Probe approval-lock exactly once per record so that both checkRecordAccessInternal
	// (called by CheckRecordAccess) and computeIsLockedWith share the same snapshot.
	// Without this, a state transition between probes could produce the racy combination
	// hasEditAccess=true with isLocked=true (or vice versa).
	approvalLocked := profile.bool(ctx, "approval_lock", func(stageCtx context.Context) bool {
		return ps.isApprovalLockedForOperation(stageCtx, record, constants.PermEdit, user)
	})

	// Record-level access checks route through checkRecordAccessWithApprovalLock so
	// isApprovalLockedForOperation is NOT re-called inside checkRecordAccessInternal.
	hasRecordRead := profile.bool(ctx, "record_read", func(stageCtx context.Context) bool {
		return ps.checkRecordAccessWithApprovalLock(stageCtx, schema, record, constants.PermRead, user, false)
	})
	hasRecordEdit := profile.bool(ctx, "record_edit", func(stageCtx context.Context) bool {
		return ps.checkRecordAccessWithApprovalLock(stageCtx, schema, record, constants.PermEdit, user, approvalLocked)
	})
	hasRecordDelete := profile.bool(ctx, "record_delete", func(stageCtx context.Context) bool {
		return ps.checkRecordAccessWithApprovalLock(stageCtx, schema, record, constants.PermDelete, user, approvalLocked)
	})
	isLocked := profile.bool(ctx, "lock_signal", func(stageCtx context.Context) bool {
		return ps.computeIsLockedWith(stageCtx, schema, record, user, approvalLocked)
	})

	return RecordAccessResult{
		HasReadAccess:   hasRecordRead,
		HasEditAccess:   hasObjEdit && hasRecordEdit,
		HasDeleteAccess: hasObjDelete && hasRecordDelete,
		IsLocked:        isLocked,
	}
}

// BatchCheckRecordAccess evaluates Read, Edit, and Delete access for multiple records
// of the same object type. For each record it calls CheckRecordAccessAll.
// Future optimization: pre-fetch manual shares and team members in batch to reduce DB queries.
func (ps *PermissionService) BatchCheckRecordAccess(
	ctx context.Context,
	schema *models.ObjectMetadata,
	records []models.SObject,
	user *models.UserSession,
) map[string]RecordAccessResult {
	ctx = withRecordAccessEvaluationCache(ctx)
	ctx = WithAccessInputCache(ctx)
	results := make(map[string]RecordAccessResult, len(records))

	// Quick exit: no records
	if len(records) == 0 {
		return results
	}

	// Pre-check object-level permissions once (shared across all records)
	objectName := ""
	if schema != nil {
		objectName = schema.APIName
	}
	if objectName != "" {
		ps.PrimeRecordAccessInputCache(ctx, objectName, records, user)
	}
	for _, rec := range records {
		recordID := ""
		if id, ok := rec[constants.FieldID]; ok {
			if idStr, ok := id.(string); ok {
				recordID = idStr
			}
		}
		if recordID == "" {
			continue
		}
		ps.primeAggregateSharingGroupMembership(ctx, schema, rec, user)

		hasObjEdit := objectName != "" && ps.CheckObjectPermissionOnRecord(ctx, schema, rec, constants.PermEdit, user)
		hasObjDelete := objectName != "" && ps.CheckObjectPermissionOnRecord(ctx, schema, rec, constants.PermDelete, user)

		// Probe approval-lock exactly once per record; reuse in all downstream calls.
		approvalLocked := ps.isApprovalLockedForOperation(ctx, rec, constants.PermEdit, user)

		hasRecordRead := ps.checkRecordAccessWithApprovalLock(ctx, schema, rec, constants.PermRead, user, false)
		hasRecordEdit := ps.checkRecordAccessWithApprovalLock(ctx, schema, rec, constants.PermEdit, user, approvalLocked)
		hasRecordDelete := ps.checkRecordAccessWithApprovalLock(ctx, schema, rec, constants.PermDelete, user, approvalLocked)

		results[recordID] = RecordAccessResult{
			HasReadAccess:   hasRecordRead,
			HasEditAccess:   hasObjEdit && hasRecordEdit,
			HasDeleteAccess: hasObjDelete && hasRecordDelete,
			IsLocked:        ps.computeIsLockedWith(ctx, schema, rec, user, approvalLocked),
		}
	}

	return results
}

func (ps *PermissionService) primeAggregateSharingGroupMembership(
	ctx context.Context,
	schema *models.ObjectMetadata,
	record models.SObject,
	user *models.UserSession,
) {
	if ps == nil || schema == nil || user == nil || user.ID == "" {
		return
	}
	distinct := map[string]struct{}{}
	ps.collectAggregateSharingGroupTargets(ctx, schema, record, distinct, map[string]struct{}{})
	if len(distinct) >= 2 {
		_, _ = ps.cachedUserGroupIDs(ctx, user.ID)
	}
}

// collectAggregateSharingGroupTargets follows the same bounded
// ControlledByParent chain that record access will evaluate and collects the
// group targets used by every object in that authority closure. This matters
// for child records such as Contact: priming only the child schema leaves the
// parent Account checks to issue one membership query per sharing rule.
//
// Parent records are loaded through the request evaluation cache, so the
// subsequent access decision reuses the exact same record snapshot. Any
// malformed edge, missing parent, or load error simply leaves the existing
// fail-closed point path in place; prefetching never changes the decision.
func (ps *PermissionService) collectAggregateSharingGroupTargets(
	ctx context.Context,
	schema *models.ObjectMetadata,
	record models.SObject,
	distinct map[string]struct{},
	visited map[string]struct{},
) {
	if ps == nil || schema == nil || len(distinct) >= 2 || ctx.Err() != nil {
		return
	}
	objectName := strings.TrimSpace(schema.APIName)
	if objectName == "" {
		return
	}
	visitKey := strings.ToLower(objectName)
	if _, seen := visited[visitKey]; seen {
		return
	}
	visited[visitKey] = struct{}{}

	for _, rule := range ps.getCachedSharingRules(ctx, objectName) {
		if rule == nil || rule.ShareWithGroupID == nil {
			continue
		}
		groupID := strings.TrimSpace(*rule.ShareWithGroupID)
		if groupID == "" {
			continue
		}
		distinct[groupID] = struct{}{}
		if len(distinct) >= 2 {
			return
		}
	}

	edge := ps.parentControlledSharingEdgeForSchema(ctx, schema, objectName)
	if edge.malformed || edge.rel == nil || record == nil {
		return
	}
	parentID, _ := record[edge.rel.FieldAPIName].(string)
	parentID = strings.TrimSpace(parentID)
	if parentID == "" || ps.metadata == nil {
		return
	}
	parentSchema := ps.metadata.GetSchema(ctx, edge.rel.ParentObjectAPIName)
	if parentSchema == nil {
		return
	}
	parentRecord, err := ps.parentRecordForAccessEvaluation(ctx, edge.rel.ParentObjectAPIName, parentID)
	if err != nil || parentRecord == nil {
		return
	}
	ps.collectAggregateSharingGroupTargets(ctx, parentSchema, parentRecord, distinct, visited)
}

// checkRecordAccessWithApprovalLock is like checkRecordAccessInternal but accepts
// a pre-computed approvalLocked value so the DB probe is not repeated.
// approvalLocked MUST have been computed for the same (record, PermEdit, user) triple
// as the caller's single probe. The pre-fetched value is forwarded directly into
// checkRecordAccessInternal via the approvalLocked *bool parameter, so the inner
// function never re-calls isApprovalLockedForOperation.
func (ps *PermissionService) checkRecordAccessWithApprovalLock(
	ctx context.Context,
	schema *models.ObjectMetadata,
	record models.SObject,
	operation string,
	user *models.UserSession,
	approvalLocked bool,
) bool {
	// Read is never approval-locked; pass nil so the inner function probes normally
	// (it will fast-return false for reads — no extra DB round-trip).
	if operation == constants.PermRead {
		return ps.checkRecordAccessInternal(ctx, schema, record, operation, user, nil, nil)
	}
	// For mutations: pass the pre-fetched approvalLocked value directly so
	// checkRecordAccessInternal skips its own probe entirely — eliminating the
	// double-probe race window the round-1 fix attempted but did not fully close.
	return ps.checkRecordAccessInternal(ctx, schema, record, operation, user, nil, &approvalLocked)
}

type parentControlledRecordAccessResult struct {
	allowed    bool
	parentOnly bool
	malformed  bool
}

func (ps *PermissionService) checkControlledByParentAccessDetailed(
	ctx context.Context,
	schema *models.ObjectMetadata,
	record models.SObject,
	operation string,
	user *models.UserSession,
	visited map[string]bool,
) parentControlledRecordAccessResult {
	if schema == nil {
		return parentControlledRecordAccessResult{}
	}
	if ps.metadata == nil {
		return parentControlledRecordAccessResult{malformed: true}
	}

	objectName := schema.APIName

	// Cycle detection — same pattern as rls_filter.go:239-243
	if visited == nil {
		visited = make(map[string]bool)
	}
	if visited[objectName] {
		return parentControlledRecordAccessResult{}
	}
	visited[objectName] = true
	defer delete(visited, objectName)

	edge := ps.parentControlledSharingEdgeForSchema(ctx, schema, objectName)
	if edge.malformed {
		return parentControlledRecordAccessResult{malformed: true}
	}
	rel := edge.rel
	if rel == nil {
		return parentControlledRecordAccessResult{}
	}

	// Extract parent FK value from the child record
	parentID, _ := record[rel.FieldAPIName].(string)
	if parentID == "" {
		return parentControlledRecordAccessResult{parentOnly: edge.parentOnly}
	}

	parentSchema := ps.metadata.GetSchema(ctx, rel.ParentObjectAPIName)
	if parentSchema == nil {
		return parentControlledRecordAccessResult{parentOnly: edge.parentOnly}
	}

	// Fetch full parent record (criteria-based sharing rules need all fields)
	parentRecord, err := ps.parentRecordForAccessEvaluation(ctx, rel.ParentObjectAPIName, parentID)
	if err != nil {
		if err == sql.ErrNoRows {
			return parentControlledRecordAccessResult{parentOnly: edge.parentOnly}
		}
		log.Warn().Err(err).
			Str("parent", rel.ParentObjectAPIName).Str("parentId", parentID).
			Msg("ControlledByParent: failed to load parent record")
		return parentControlledRecordAccessResult{parentOnly: edge.parentOnly}
	}

	// Recursively check access on the parent record (nil = probe approval-lock fresh for the parent record)
	parentOperation := parentControlledParentCapabilityOperation(operation)
	return parentControlledRecordAccessResult{
		allowed:    ps.checkRecordAccessInternal(ctx, parentSchema, parentRecord, parentOperation, user, visited, nil),
		parentOnly: edge.parentOnly,
	}
}