package permissions

import (
	"context"
	"fmt"
	"log"
	"strings"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"

	"github.com/nexuscrm/kernel/pkg/formula"
	domainrls "github.com/nexuscrm/kernel/pkg/rls"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// rlsCriteriaCompileErrorTotal counts sharing-rule criteria that failed to compile to SQL.
// Labeled by (object, error_class). The caller silently `continue`s past the failing rule,
// which DENIES the user records this rule would have granted (NexusCRM RLS is additive-grant).
// Issue #949 PR4 (Path B, B6c observability) — zero behavior change.
var rlsCriteriaCompileErrorTotal = promauto.NewCounterVec(
	prometheus.CounterOpts{
		Name: "nexuscrm_rls_criteria_compile_error_total",
		Help: "RLS sharing-rule criteria that failed to compile to SQL — incremented per (object, error_class). " +
			"Caller path silently `continue`s past the failing rule, which DENIES the user records this rule would have granted " +
			"(NexusCRM RLS is additive). Issue #949 PR4 (B6c observability).",
	},
	[]string{"object", "error_class"},
)

// BuildRLSResult returns pre-computed RLS data for a (user, object) pair.
// This is the SINGLE source of truth for query-level RLS — all query paths
// must call this function.
//
// Implements the same 7 access layers as CheckRecordAccess:
//
//  0. SuperUser bypass
//  1. OWD (PublicReadWrite / PublicRead)
//  2. ViewAllRecords / ModifyAllRecords
//  3. Ownership (self + subordinates via role hierarchy)
//  4. Group/Queue ownership
//  5. Sharing rules (criteria-less: resolved to user IDs; criteria-based: deferred)
//  6. Manual share + Team member (via _System_RecordShare + _System_TeamMember)
func (ps *PermissionService) BuildRLSResult(
	ctx context.Context,
	schema *models.ObjectMetadata,
	objectName string,
	user *models.UserSession,
) *RLSResult {
	return ps.buildRLSResult(ctx, schema, objectName, user, make(map[string]bool))
}

func (ps *PermissionService) buildRLSResult(
	ctx context.Context,
	schema *models.ObjectMetadata,
	objectName string,
	user *models.UserSession,
	visited map[string]bool,
) *RLSResult {
	denyAll := &RLSResult{ObjectAPIName: objectName, DenyAll: true}

	// nil schema → deny-all
	if schema == nil {
		return denyAll
	}

	// nil user → deny-all
	if user == nil {
		return denyAll
	}

	visitKey := strings.ToLower(strings.TrimSpace(objectName))
	if visitKey != "" {
		if visited[visitKey] {
			return denyAll
		}
		visited[visitKey] = true
		defer delete(visited, visitKey)
	}

	// Delegated row-read policies are still a row-shape constraint for admins:
	// query adapters need the target filter to preserve /query vs /queryAll
	// deleted-target semantics, even when the target RLS itself is unrestricted.
	// Metadata may instead declare target visibility as an additive grant to the
	// source object's normal row visibility (for example, a polymorphic activity
	// that remains visible to its owner when it has no target record).
	var additiveTargetFilters []domainrls.TargetRecordFilter
	if policy := schema.RowReadPolicy(); policy != nil && strings.EqualFold(strings.TrimSpace(policy.Kind), "targetRecordRead") {
		filters := ps.buildTargetRecordReadFilters(ctx, schema, user, visited)
		if !policy.IncludeSourceVisibility {
			if len(filters) == 0 {
				return denyAll
			}
			return &RLSResult{ObjectAPIName: objectName, TargetRecordFilters: filters}
		}
		additiveTargetFilters = filters
	}

	// Layer 0: super-user capability bypass.
	if ps.HasSuperUserCapability(ctx, user) {
		return &RLSResult{Skip: true, ObjectAPIName: objectName}
	}

	// L0-D D1: dispatch on the app-blind row-visibility classification instead of
	// branching on constants.SharingModel* directly. The SharingModel switch is
	// confined to models.RowVisibilityFor.
	rowVis := models.RowVisibilityFor(schema)

	// Layer 0.5/0.6: FolderScoped — dedicated folder access control via
	// FolderAccessService. Folders use FolderShare metadata instead of standard
	// RLS. VisibleFolderIDs returns the pre-computed set of folder IDs the user
	// can access; nil means "all visible." FolderScopeField selects the filtered
	// column: "" = the record Id (FolderSharing); "FolderId" = the row's folder
	// reference (FolderContentSharing — Report/Dashboard access derived from folder).
	if rowVis.Kind == models.RowVisibilityFolderScoped {
		folderResult := &RLSResult{
			ObjectAPIName:       objectName,
			TargetRecordFilters: additiveTargetFilters,
		}
		if ps.folderAccess == nil {
			// FolderAccessService not wired — deny all (fail-closed)
			if len(additiveTargetFilters) > 0 {
				return folderResult
			}
			return denyAll
		}
		ids, err := ps.folderAccess.VisibleFolderIDs(ctx, user)
		if err != nil {
			// Preserve the original per-sub-case logging: FolderSharing (record-Id
			// scope) logged; FolderContentSharing (FolderId scope) did not.
			if rowVis.FolderScopeField == "" {
				log.Printf("[RLS] FolderSharing: VisibleFolderIDs error for user %s: %v", user.ID, err)
			}
			if len(additiveTargetFilters) > 0 {
				return folderResult
			}
			return denyAll
		}
		if ids == nil {
			// nil = all folders visible (SuperUser/ViewAll handled inside VisibleFolderIDs)
			return &RLSResult{Skip: true, ObjectAPIName: objectName}
		}
		folderResult.AllowedIDs = ids
		folderResult.AllowedIDsColumn = rowVis.FolderScopeField
		return folderResult
	}

	if policy := schema.RowReadPolicy(); isOwnerVisibilityRowReadPolicy(policy) {
		if ps.CheckSystemPermission(ctx, user, string(constants.PermViewAllData)) ||
			ps.CheckSystemPermission(ctx, user, string(constants.PermModifyAllData)) {
			return &RLSResult{Skip: true, ObjectAPIName: objectName}
		}
		if rowVis.Kind != models.RowVisibilityParentControlled && !schema.UsesDerivedAccessPolicy() {
			objPerm, err := ps.loadEffectiveObjectPermission(ctx, user, objectName)
			if err == nil && objPerm != nil {
				if objPerm.PermissionsViewAllRecords || objPerm.PermissionsModifyAllRecords {
					return &RLSResult{Skip: true, ObjectAPIName: objectName}
				}
			}
		}
		filter := buildOwnerVisibilityFilter(policy, user)
		if filter == nil {
			return denyAll
		}
		return &RLSResult{ObjectAPIName: objectName, OwnerVisibility: filter}
	}

	// Layer 1: OWD — PublicReadWrite / PublicRead (read is always the query operation)
	if rowVis.Kind == models.RowVisibilityPublicRead ||
		rowVis.Kind == models.RowVisibilityPublicReadWrite {
		return &RLSResult{Skip: true, ObjectAPIName: objectName}
	}

	result := &RLSResult{
		ObjectAPIName:       objectName,
		TargetRecordFilters: additiveTargetFilters,
	}

	var parentControlledEdge parentControlledSharingEdge
	isParentControlled := rowVis.Kind == models.RowVisibilityParentControlled
	// ControlledByParent adds parent-derived visibility. Required/master-detail
	// sharing-parent edges are fully inherited and suppress owner/share fallback;
	// optional edges retain owner/share fallback for standalone/orphan records.
	if isParentControlled {
		parentControlledEdge = ps.parentControlledSharingEdgeForSchema(ctx, schema, objectName)
		if parentControlledEdge.malformed {
			return denyAll
		}
		result.ParentFilter = ps.buildControlledByParentFilterWithGuard(ctx, schema, objectName, user, visited)
		if result.ParentFilter != nil {
			result.ParentFilter.ParentDerivedOnly = parentControlledEdge.parentOnly
		}
	}

	// Layer 2a: System-level ViewAllData / ModifyAllData (global, all objects)
	if ps.CheckSystemPermission(ctx, user, string(constants.PermViewAllData)) ||
		ps.CheckSystemPermission(ctx, user, string(constants.PermModifyAllData)) {
		return &RLSResult{Skip: true, ObjectAPIName: objectName}
	}

	// Layer 2b: Object-level ViewAllRecords / ModifyAllRecords (per-object)
	// applies only to self-owned object-permission surfaces. Derived-policy
	// objects delegate ownership to AccessPolicy + parent visibility, so stale
	// self OP rows must not short-circuit query RLS.
	if !isParentControlled && !schema.UsesDerivedAccessPolicy() {
		objPerm, err := ps.loadEffectiveObjectPermission(ctx, user, objectName)
		if err == nil && objPerm != nil {
			if objPerm.PermissionsViewAllRecords || objPerm.PermissionsModifyAllRecords {
				return &RLSResult{Skip: true, ObjectAPIName: objectName}
			}
		}
	}

	if isParentControlled && parentControlledEdge.parentOnly {
		return result
	}

	// Layer 3: Check HasOwnerField
	hasOwner := false
	for _, f := range schema.Fields {
		if f.APIName == constants.FieldOwnerID {
			hasOwner = true
			break
		}
	}

	result.HasOwnerField = hasOwner
	result.UserID = user.ID
	result.IncludeManualShares = true
	result.IncludeTeamMembers = true

	if user.ID != "" {
		if groupIDs, ok := ps.cachedUserGroupIDs(ctx, user.ID); ok && len(groupIDs) > 0 {
			result.GroupOwnerIDs = groupIDs
		} else {
			groupIDs, err := ps.repo.GetUserGroupIDs(ctx, user.ID)
			if err == nil && len(groupIDs) > 0 {
				result.GroupOwnerIDs = groupIDs
			}
		}

		// issue #2585: drop the manual-share / team-member semi-join arms when the
		// user provably has NONE for this object. An `Id IN (SELECT … _System_RecordShare
		// / _System_TeamMember …)` semi-join cannot combine with an ordered-index TopN,
		// so its mere presence forces a full table scan + sort of the entire visible
		// set — even when it grants zero rows (the common case for broad-visibility
		// users whose access comes from owner/role/group/criteria, not manual shares).
		//
		// Dropping a PROVEN-EMPTY arm is over-grant-safe: it can only under-grant on a
		// concurrent share/team ADD (the safe direction), never over-grant on a revoke.
		// A NON-empty arm keeps its subquery, which the database evaluates at SELECT
		// time, so a concurrent revoke is reflected exactly as before. On a lookup
		// error the arm is kept (fail-safe). The existence probes are LIMIT 1.
		if ids, err := ps.repo.GetManualShareRecordIDs(ctx, objectName, user.ID); err == nil && len(ids) == 0 {
			result.IncludeManualShares = false
		}
		if ids, err := ps.repo.GetTeamMemberRecordIDs(ctx, objectName, user.ID); err == nil && len(ids) == 0 {
			result.IncludeTeamMembers = false
		}
	}

	if hasOwner {
		// Self
		ownerIDs := []string{user.ID}

		// Subordinates via role hierarchy (only if EnableHierarchySharing)
		if rowVis.HierarchyGranted && user.RoleID != nil && *user.RoleID != "" {
			subordinateIDs, err := ps.GetSubordinateUserIDs(ctx, *user.RoleID)
			if err == nil {
				ownerIDs = append(ownerIDs, subordinateIDs...)
			}
		}

		result.OwnerUserIDs = ownerIDs
	}

	// Layer 5: Sharing rules
	rules := ps.getCachedSharingRules(ctx, objectName)
	sharingRuleOwnerIDs := make(map[string]bool)
	for _, rule := range rules {
		// Check if user matches the ShareWith target of this rule
		userMatchesTarget := ps.userMatchesSharingRuleTarget(ctx, user, rule)
		if !userMatchesTarget {
			continue
		}

		hasCriteria := rule.Criteria != "" && rule.Criteria != "[]"

		if hasCriteria {
			criteriaExpr, criteriaErr := SharingRuleRuntimeCriteriaExpression(rule.Criteria, schema)
			if criteriaErr != nil {
				errClass := classifyFormulaCompileError(criteriaErr)
				rlsCriteriaCompileErrorTotal.WithLabelValues(objectName, errClass).Inc()
				log.Printf("[RLS] criteria compile failed rule=%s object=%s error_class=%s err=%v - user denied records this rule would grant (additive-grant model)", rule.ID, objectName, errClass, criteriaErr)
				continue
			}
			// Criteria-based rule: compile the formula expression to a SQL
			// WHERE fragment so it applies at query time (not just single-record).
			// Use API names (no field mapper) because:
			// 1. Formula views (v_{object}) expose columns as API names
			// 2. Base tables use API names for standard fields (PhysicalName is empty)
			// 3. This matches existing RLS predicates (OwnerId, Id) which are API names
			sqlFrag, args, err := formula.ToSQL(criteriaExpr)
			if err != nil {
				errClass := classifyFormulaCompileError(err)
				rlsCriteriaCompileErrorTotal.WithLabelValues(objectName, errClass).Inc()
				log.Printf("[RLS] criteria compile failed rule=%s object=%s error_class=%s err=%v — user denied records this rule would grant (additive-grant model)", rule.ID, objectName, errClass, err)
				continue
			}
			// Resolve parameterized placeholders into a self-contained SQL string,
			// consistent with how the rest of the RLS system embeds values.
			resolved := resolveParams(sqlFrag, args)
			if resolved != "" {
				result.CriteriaSQLFragments = append(result.CriteriaSQLFragments, resolved)
				result.CriteriaExpressions = append(result.CriteriaExpressions, criteriaExpr)
			}
			continue
		}

		// Criteria-less rule: this rule shares ALL records from the SharedFrom source.
		// The SharedFrom source is a role or group — collect all user IDs from it.
		if rule.SharedFromRoleID != nil && *rule.SharedFromRoleID != "" {
			// All users in this role (and below, since role hierarchy implies inclusion)
			roleUserIDs, err := ps.userRepo.GetUserIDsByRoleIDs(ctx, []string{*rule.SharedFromRoleID})
			if err == nil {
				for _, id := range roleUserIDs {
					sharingRuleOwnerIDs[id] = true
				}
			}
			// Also include subordinates of that role
			subordinateRoleIDs := ps.GetRoleDescendants(*rule.SharedFromRoleID)
			if len(subordinateRoleIDs) > 0 {
				subUserIDs, err := ps.userRepo.GetUserIDsByRoleIDs(ctx, subordinateRoleIDs)
				if err == nil {
					for _, id := range subUserIDs {
						sharingRuleOwnerIDs[id] = true
					}
				}
			}
		}
		if rule.SharedFromGroupID != nil && *rule.SharedFromGroupID != "" {
			// Records owned by members of this group are visible via this sharing rule.
			// Store the group ID; the SQL adapter will produce a subquery:
			// OwnerId IN (SELECT MemberId FROM _System_GroupMembershipClosure WHERE GroupId = '<groupID>')
			result.SharingRuleFromGroupIDs = append(result.SharingRuleFromGroupIDs, *rule.SharedFromGroupID)
		}
	}

	if len(sharingRuleOwnerIDs) > 0 {
		ids := make([]string, 0, len(sharingRuleOwnerIDs))
		for id := range sharingRuleOwnerIDs {
			ids = append(ids, id)
		}
		result.SharingRuleOwnerIDs = ids
	}

	// issue #2585: same over-grant-safe drop for the group-from-owner arm — if none
	// of the shared-from groups has any member, the `OwnerId IN (SELECT MemberId FROM
	// closure …)` semi-join grants nothing, so omit it (keeps the predicate sargable).
	// Non-empty → kept as the SELECT-time subquery; lookup error → kept (fail-safe).
	if len(result.SharingRuleFromGroupIDs) > 0 {
		if ids, err := ps.repo.GetGroupMemberIDs(ctx, result.SharingRuleFromGroupIDs); err == nil && len(ids) == 0 {
			result.SharingRuleFromGroupIDs = nil
		}
	}

	return result
}

func (ps *PermissionService) buildTargetRecordReadFilters(
	ctx context.Context,
	schema *models.ObjectMetadata,
	user *models.UserSession,
	visited map[string]bool,
) []domainrls.TargetRecordFilter {
	policy := schema.RowReadPolicy()
	if policy == nil || !strings.EqualFold(strings.TrimSpace(policy.Kind), "targetRecordRead") {
		return nil
	}
	if ps.metadata == nil {
		return nil
	}
	associations, valid := policy.TargetRecordReadAssociations()
	if !valid {
		return nil
	}
	if len(policy.TargetAssociations) == 0 {
		if len(associations) != 1 {
			return nil
		}
		targetObjectField := associations[0].TargetObjectField
		targetRecordField := associations[0].TargetRecordField
		var legacyFilters []domainrls.TargetRecordFilter
		for _, targetSchema := range ps.metadata.GetSchemas(ctx) {
			if !targetRecordReadAllowsTarget(schema, targetRecordField, targetSchema.APIName) ||
				!ps.canReadTargetRecordReadSchema(ctx, schema, targetSchema, user) {
				continue
			}
			legacyFilters = append(legacyFilters, domainrls.TargetRecordFilter{
				TargetObjectAPIName: targetSchema.APIName,
				TargetObjectField:   targetObjectField,
				TargetRecordField:   targetRecordField,
				TargetRLSResult:     ps.buildRLSResult(ctx, targetSchema, targetSchema.APIName, user, visited),
				TargetSchema:        targetSchema,
			})
		}
		return legacyFilters
	}
	var filters []domainrls.TargetRecordFilter
	for _, association := range associations {
		targetObjectField := association.TargetObjectField
		targetRecordField := association.TargetRecordField
		filter := domainrls.TargetRecordFilter{
			TargetObjectField: targetObjectField,
			TargetRecordField: targetRecordField,
			Alternatives:      make([]domainrls.TargetRecordFilterAlternative, 0),
		}
		for _, targetSchema := range ps.metadata.GetSchemas(ctx) {
			if !targetRecordReadAllowsTarget(schema, targetRecordField, targetSchema.APIName) {
				continue
			}
			if !ps.canReadTargetRecordReadSchema(ctx, schema, targetSchema, user) {
				continue
			}
			targetResult := ps.buildRLSResult(ctx, targetSchema, targetSchema.APIName, user, visited)
			filter.Alternatives = append(filter.Alternatives, domainrls.TargetRecordFilterAlternative{
				TargetObjectAPIName: targetSchema.APIName,
				TargetRLSResult:     targetResult,
				TargetSchema:        targetSchema,
			})
		}
		if len(filter.Alternatives) == 1 {
			alternative := filter.Alternatives[0]
			filter.TargetObjectAPIName = alternative.TargetObjectAPIName
			filter.TargetRLSResult = alternative.TargetRLSResult
			filter.TargetSchema = alternative.TargetSchema
		}
		filters = append(filters, filter)
	}
	return filters
}

func targetRecordReadAllowsTarget(sourceSchema *models.ObjectMetadata, targetRecordField, targetObjectAPIName string) bool {
	field := models.FindField(sourceSchema, targetRecordField)
	if field == nil || len(field.ReferenceTo) == 0 {
		return true
	}
	for _, allowed := range field.ReferenceTo {
		if strings.EqualFold(strings.TrimSpace(allowed), strings.TrimSpace(targetObjectAPIName)) {
			return true
		}
	}
	return false
}

func (ps *PermissionService) canReadTargetRecordReadSchema(ctx context.Context, sourceSchema *models.ObjectMetadata, targetSchema *models.ObjectMetadata, user *models.UserSession) bool {
	if sourceSchema == nil || targetSchema == nil || strings.TrimSpace(targetSchema.APIName) == "" {
		return false
	}
	if strings.EqualFold(targetSchema.APIName, sourceSchema.APIName) {
		return false
	}
	if ps.IsBlockedObject(ctx, targetSchema.APIName, constants.PermRead) {
		return false
	}
	if ps.CheckObjectPermissionWithUser(ctx, targetSchema.APIName, constants.PermRead, user) {
		return true
	}
	return ps.CheckSystemPermission(ctx, user, string(constants.PermViewAllData)) ||
		ps.CheckSystemPermission(ctx, user, string(constants.PermModifyAllData))
}

func (ps *PermissionService) buildControlledByParentFilterWithGuard(
	ctx context.Context,
	schema *models.ObjectMetadata,
	objectName string,
	user *models.UserSession,
	visited map[string]bool,
) *ParentRLSFilter {
	if ps.metadata == nil {
		return nil
	}

	rel := ps.metadata.GetSharingParentRelationship(ctx, objectName)
	if rel == nil {
		return nil
	}

	parentObjectName := rel.ParentObjectAPIName
	parentSchema := ps.metadata.GetSchema(ctx, parentObjectName)
	if parentSchema == nil {
		return nil
	}
	parentResult := ps.buildRLSResult(ctx, parentSchema, parentObjectName, user, visited)

	return &ParentRLSFilter{
		ParentObjectAPIName: parentObjectName,
		ForeignKeyField:     rel.FieldAPIName,
		ParentRLSResult:     parentResult,
		ParentSchema:        parentSchema,
	}
}

// userMatchesSharingRuleTarget checks if the user is the target (ShareWith) of a sharing rule.
func (ps *PermissionService) userMatchesSharingRuleTarget(ctx context.Context, user *models.UserSession, rule *models.SharingRule) bool {
	if rule.ShareWithRoleID != nil && *rule.ShareWithRoleID != "" {
		if ps.isUserInRoleOrAbove(user.RoleID, rule.ShareWithRoleID) {
			return true
		}
	}
	if rule.ShareWithGroupID != nil && *rule.ShareWithGroupID != "" {
		// Session user's membership: served from request-scoped point memoization
		// (issue #1585) so repeated target checks reuse the exact pair answer
		// without a full user-group preload.
		if ps.isUserInGroupForSession(ctx, *rule.ShareWithGroupID, user.ID, user.ID) {
			return true
		}
	}
	return false
}

// RLSResultToFilterExpr converts an RLSResult into a SQL WHERE fragment.
// tableQualifier is the physical FROM target (view name, alias, or table name).
// Pass "" to default to the logical ObjectAPIName.
func RLSResultToFilterExpr(result *RLSResult, tableQualifier string) string {
	return domainrls.ToSQLFilter(result, tableQualifier)
}

// classifyFormulaCompileError buckets formula.ToSQL errors into stable
// label values for the rlsCriteriaCompileErrorTotal metric.
// Categories: "parse" (lexer/parser failure), "unknown_function"
// (function not in compiler's allowlist), "arity" (function called with
// wrong number of args), "other" (everything else).
func classifyFormulaCompileError(err error) string {
	if err == nil {
		return "none"
	}
	msg := strings.ToLower(err.Error())
	switch {
	case strings.Contains(msg, "parse") || strings.Contains(msg, "syntax"):
		return "parse"
	case strings.Contains(msg, "unknown function") || strings.Contains(msg, "not supported") || strings.Contains(msg, "unsupported"):
		return "unknown_function"
	case strings.Contains(msg, "arity") || strings.Contains(msg, "argument") || strings.Contains(msg, "arg count"):
		return "arity"
	default:
		return "other"
	}
}

// resolveParams substitutes ? placeholders with their actual values to produce
// a self-contained SQL fragment, consistent with how the RLS system embeds values.
func resolveParams(sql string, args []interface{}) string {
	if len(args) == 0 || !strings.Contains(sql, "?") {
		return sql
	}
	var b strings.Builder
	b.Grow(len(sql))
	argIdx := 0
	for _, r := range sql {
		if r == '?' && argIdx < len(args) {
			b.WriteString(rlsSQLLiteral(args[argIdx]))
			argIdx++
			continue
		}
		b.WriteRune(r)
	}
	return b.String()
}

func rlsSQLLiteral(arg interface{}) string {
	switch v := arg.(type) {
	case string:
		return fmt.Sprintf("'%s'", strings.ReplaceAll(v, "'", "''"))
	case int:
		return fmt.Sprintf("%d", v)
	case int64:
		return fmt.Sprintf("%d", v)
	case float64:
		return fmt.Sprintf("%v", v)
	case bool:
		if v {
			return "1"
		}
		return "0"
	default:
		return fmt.Sprintf("'%s'", strings.ReplaceAll(fmt.Sprintf("%v", v), "'", "''"))
	}
}