package permissions

import (
	"context"
	"errors"
	"sort"
	"strings"
	"sync"
	"time"

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

var (
	errBoundedFieldPermissionProjectionUnavailable = errors.New("bounded effective field permission projection is unavailable")
	// ErrFieldPermissionProjectionOverloaded is returned when both the active
	// object-projection slots and the finite FIFO waiter queue are full. Callers
	// must propagate it as a retriable service failure, never as a permission
	// denial or an omitted field.
	ErrFieldPermissionProjectionOverloaded = &FieldPermissionProjectionOverloadedError{}
)

const (
	defaultBoundedFieldPermissionProjectionLimit       = 32
	defaultBoundedFieldPermissionProjectionWaiterLimit = 32
	maxBoundedFieldPermissionProjectionLimit           = 1024
	maxBoundedFieldPermissionProjectionWaiterLimit     = 4096
)

// FieldPermissionProjectionOverloadedError is the typed load-shed signal for
// bounded FLS projection admission.
type FieldPermissionProjectionOverloadedError struct{}

func (*FieldPermissionProjectionOverloadedError) Error() string {
	return "bounded effective field permission projection admission is full"
}

func (*FieldPermissionProjectionOverloadedError) RetryAfterSeconds() int {
	return 1
}

func (*FieldPermissionProjectionOverloadedError) RetriableCondition() bool {
	return true
}

func (*FieldPermissionProjectionOverloadedError) InternalCondition() bool {
	return true
}

type fieldProjectionWaiter struct {
	ready   chan struct{}
	granted bool
}

type boundedFieldProjectionAdmission struct {
	mu          sync.Mutex
	active      int
	activeLimit int
	waiterLimit int
	waiters     []*fieldProjectionWaiter
}

func newBoundedFieldProjectionAdmission(activeLimit, waiterLimit int) *boundedFieldProjectionAdmission {
	if activeLimit <= 0 || activeLimit > maxBoundedFieldPermissionProjectionLimit {
		activeLimit = defaultBoundedFieldPermissionProjectionLimit
	}
	if waiterLimit < 0 || waiterLimit > maxBoundedFieldPermissionProjectionWaiterLimit {
		waiterLimit = defaultBoundedFieldPermissionProjectionWaiterLimit
	}
	return &boundedFieldProjectionAdmission{activeLimit: activeLimit, waiterLimit: waiterLimit}
}

func (a *boundedFieldProjectionAdmission) release() {
	a.mu.Lock()
	defer a.mu.Unlock()
	if len(a.waiters) == 0 {
		if a.active > 0 {
			a.active--
		}
		return
	}
	waiter := a.waiters[0]
	copy(a.waiters, a.waiters[1:])
	a.waiters[len(a.waiters)-1] = nil
	a.waiters = a.waiters[:len(a.waiters)-1]
	waiter.granted = true
	close(waiter.ready)
}

func (a *boundedFieldProjectionAdmission) snapshot() (active, waiters int) {
	if a == nil {
		return 0, 0
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.active, len(a.waiters)
}

func fieldProjectionReleaseOnce(a *boundedFieldProjectionAdmission) func() {
	var once sync.Once
	return func() { once.Do(a.release) }
}

func (a *boundedFieldProjectionAdmission) acquire(ctx context.Context) (func(), error) {
	a.mu.Lock()
	if a.active < a.activeLimit && len(a.waiters) == 0 {
		a.active++
		a.mu.Unlock()
		return fieldProjectionReleaseOnce(a), nil
	}
	if len(a.waiters) >= a.waiterLimit {
		a.mu.Unlock()
		return nil, ErrFieldPermissionProjectionOverloaded
	}
	waiter := &fieldProjectionWaiter{ready: make(chan struct{})}
	a.waiters = append(a.waiters, waiter)
	a.mu.Unlock()

	select {
	case <-waiter.ready:
		if err := ctx.Err(); err != nil {
			a.release()
			return nil, err
		}
		return fieldProjectionReleaseOnce(a), nil
	case <-ctx.Done():
		a.mu.Lock()
		if waiter.granted {
			a.mu.Unlock()
			a.release()
			return nil, ctx.Err()
		}
		for i, queued := range a.waiters {
			if queued != waiter {
				continue
			}
			copy(a.waiters[i:], a.waiters[i+1:])
			a.waiters[len(a.waiters)-1] = nil
			a.waiters = a.waiters[:len(a.waiters)-1]
			break
		}
		a.mu.Unlock()
		return nil, ctx.Err()
	}
}

func (ps *PermissionService) acquireBoundedFieldPermissionProjection(ctx context.Context) (func(), error) {
	if ps == nil {
		return nil, errBoundedFieldPermissionProjectionUnavailable
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	ps.fieldProjectionAdmissionOnce.Do(func() {
		if ps.fieldProjectionAdmission == nil {
			ps.fieldProjectionAdmission = newBoundedFieldProjectionAdmission(
				defaultBoundedFieldPermissionProjectionLimit,
				defaultBoundedFieldPermissionProjectionWaiterLimit,
			)
		}
	})
	release, err := ps.fieldProjectionAdmission.acquire(ctx)
	if err != nil {
		recordFieldPermissionBulkRejection(ctx)
		return nil, err
	}
	return release, nil
}

// ListEffectiveObjectPermissionsForUser retrieves all effective object permissions for a user (Profile + Permission Sets).
// Parent-controlled children are read-time projections of the runtime CRUD
// decision: their ObjectPermissions rows are not authoritative, so the effective
// list must expose the same answer that CheckObjectPermissionWithUser enforces.
func (ps *PermissionService) ListEffectiveObjectPermissionsForUser(ctx context.Context, userID, profileID string) ([]models.ObjectPermissions, error) {
	start := time.Now()
	rows := 0
	requestID, _ := ctx.Value(constants.ContextKeyRequestID).(string)
	defer func() {
		recordObjectPermissionBulkLoad(ctx, rows, time.Since(start))
	}()

	user := &models.UserSession{ID: userID, ProfileID: profileID}
	if ps.HasSuperUserCapability(ctx, user) && ps != nil && ps.metadata != nil && len(ps.metadata.GetSchemas(ctx)) > 0 {
		superStart := time.Now()
		perms := ps.projectSuperUserEffectiveObjectPermissions(ctx, userID, profileID)
		rows = len(perms)
		logObjectPermBulkPhase(requestID, userID, profileID, "superuser_projection", len(perms), superStart)
		return perms, nil
	}

	queryStart := time.Now()
	perms, err := ps.repo.ListEffectiveObjectPermissionsForUser(ctx, userID, profileID)
	logObjectPermBulkPhase(requestID, userID, profileID, "repo_query", len(perms), queryStart)
	if err != nil {
		return nil, err
	}
	projectStart := time.Now()
	projected := ps.projectParentControlledEffectiveObjectPermissions(ctx, userID, profileID, perms)
	logObjectPermBulkPhase(requestID, userID, profileID, "parent_controlled_projection", len(projected), projectStart)
	rows = len(projected)
	return projected, nil
}

// ListEffectiveObjectPermissionsForUserResources evaluates the authoritative
// CRUD decision only for the requested neutral Catalog resources. It reuses the
// request-scoped permission input cache, so repeated operations and
// parent-controlled delegation do not repeat point reads. The complete
// administrative projection remains ListEffectiveObjectPermissionsForUser.
func (ps *PermissionService) ListEffectiveObjectPermissionsForUserResources(
	ctx context.Context,
	userID string,
	profileID string,
	resourceTypes []string,
) ([]models.ObjectPermissions, error) {
	if ps == nil || ps.metadata == nil || len(resourceTypes) == 0 {
		return nil, nil
	}
	if ctx == nil {
		ctx = context.Background()
	}
	ctx = WithAccessInputCache(ctx)
	user := &models.UserSession{ID: userID, ProfileID: profileID}
	projected := make([]models.ObjectPermissions, 0, len(resourceTypes))
	for _, resourceType := range resourceTypes {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		schema := ps.metadata.GetSchema(ctx, resourceType)
		if schema == nil {
			continue
		}
		permission := ps.runtimeProjectedObjectPermission(ctx, schema.APIName, user)
		if objectPermissionHasAnyGrant(permission) {
			projected = append(projected, permission)
		}
	}
	sort.Slice(projected, func(i, j int) bool {
		return projected[i].SobjectType < projected[j].SobjectType
	})
	return projected, nil
}

func logObjectPermBulkPhase(requestID, userID, profileID, phase string, rows int, started time.Time) {
	zlog.Info().
		Str("event", "obj_perm_bulk_phase_timing").
		Str("request_id", requestID).
		Str("user_id", userID).
		Str("profile_id", profileID).
		Str("phase", phase).
		Int("rows", rows).
		Int64("elapsed_ms", time.Since(started).Milliseconds()).
		Msg("Object permission bulk phase timing")
}

func (ps *PermissionService) projectSuperUserEffectiveObjectPermissions(ctx context.Context, userID, profileID string) []models.ObjectPermissions {
	if ps == nil || ps.metadata == nil {
		return nil
	}

	user := &models.UserSession{ID: userID, ProfileID: profileID}
	schemas := ps.metadata.GetSchemas(ctx)
	projected := make([]models.ObjectPermissions, 0, len(schemas))
	for _, schema := range schemas {
		if schema == nil {
			continue
		}
		perm := ps.runtimeProjectedObjectPermission(ctx, schema.APIName, user)
		if objectPermissionHasAnyGrant(perm) {
			projected = append(projected, perm)
		}
	}

	sort.Slice(projected, func(i, j int) bool {
		return strings.ToLower(projected[i].SobjectType) < strings.ToLower(projected[j].SobjectType)
	})
	return projected
}

func (ps *PermissionService) projectParentControlledEffectiveObjectPermissions(
	ctx context.Context,
	userID string,
	profileID string,
	perms []models.ObjectPermissions,
) []models.ObjectPermissions {
	if ps == nil || ps.metadata == nil {
		return perms
	}

	schemas := ps.metadata.GetSchemas(ctx)
	projected := append([]models.ObjectPermissions(nil), perms...)
	projected = ps.suppressBlockedGenericDeletePermissions(projected, schemas)
	indexByObject := make(map[string]int, len(projected))
	for i := range projected {
		key := strings.ToLower(strings.TrimSpace(projected[i].SobjectType))
		if key == "" {
			continue
		}
		indexByObject[key] = i
	}

	user := &models.UserSession{ID: userID, ProfileID: profileID}
	for _, schema := range schemas {
		if schema == nil {
			continue
		}
		if schema.SharingModel != constants.SharingModelControlledByParent {
			continue
		}
		if ps.metadata.GetParentControlledObject(ctx, schema) == "" {
			key := strings.ToLower(schema.APIName)
			if idx, exists := indexByObject[key]; exists {
				projected = removeObjectPermissionAt(projected, idx)
				indexByObject = rebuildObjectPermissionIndex(projected)
			}
			continue
		}

		perm := ps.runtimeProjectedObjectPermission(ctx, schema.APIName, user)
		key := strings.ToLower(schema.APIName)
		idx, exists := indexByObject[key]
		if !objectPermissionHasAnyGrant(perm) {
			if exists {
				projected = removeObjectPermissionAt(projected, idx)
				indexByObject = rebuildObjectPermissionIndex(projected)
			}
			continue
		}
		if exists {
			projected[idx] = perm
			continue
		}
		indexByObject[key] = len(projected)
		projected = append(projected, perm)
	}
	return projected
}

func (ps *PermissionService) suppressBlockedGenericDeletePermissions(perms []models.ObjectPermissions, schemas []*models.ObjectMetadata) []models.ObjectPermissions {
	if ps == nil {
		return perms
	}
	blockedObjects := make(map[string]struct{})
	for _, schema := range schemas {
		if schemaBlocksGenericDelete(schema) {
			blockedObjects[strings.ToLower(schema.APIName)] = struct{}{}
		}
	}
	out := perms[:0]
	for _, perm := range perms {
		if _, blocked := blockedObjects[strings.ToLower(perm.SobjectType)]; blocked {
			perm.PermissionsDelete = false
		}
		if objectPermissionHasAnyGrant(perm) {
			out = append(out, perm)
		}
	}
	return out
}

func (ps *PermissionService) runtimeProjectedObjectPermission(ctx context.Context, objectAPIName string, user *models.UserSession) models.ObjectPermissions {
	return models.ObjectPermissions{
		SobjectType:                 objectAPIName,
		PermissionsRead:             ps.CheckObjectPermissionWithUser(ctx, objectAPIName, constants.PermRead, user),
		PermissionsCreate:           ps.CheckObjectPermissionWithUser(ctx, objectAPIName, constants.PermCreate, user),
		PermissionsEdit:             ps.CheckObjectPermissionWithUser(ctx, objectAPIName, constants.PermEdit, user),
		PermissionsDelete:           ps.CheckObjectPermissionWithUser(ctx, objectAPIName, constants.PermDelete, user),
		PermissionsViewAllRecords:   ps.CheckObjectPermissionWithUser(ctx, objectAPIName, "ViewAllRecords", user),
		PermissionsModifyAllRecords: ps.CheckObjectPermissionWithUser(ctx, objectAPIName, "ModifyAllRecords", user),
	}
}

func objectPermissionHasAnyGrant(perm models.ObjectPermissions) bool {
	return perm.PermissionsRead ||
		perm.PermissionsCreate ||
		perm.PermissionsEdit ||
		perm.PermissionsDelete ||
		perm.PermissionsViewAllRecords ||
		perm.PermissionsModifyAllRecords
}

func removeObjectPermissionAt(perms []models.ObjectPermissions, idx int) []models.ObjectPermissions {
	if idx < 0 || idx >= len(perms) {
		return perms
	}
	copy(perms[idx:], perms[idx+1:])
	return perms[:len(perms)-1]
}

func rebuildObjectPermissionIndex(perms []models.ObjectPermissions) map[string]int {
	indexByObject := make(map[string]int, len(perms))
	for i := range perms {
		key := strings.ToLower(strings.TrimSpace(perms[i].SobjectType))
		if key != "" {
			indexByObject[key] = i
		}
	}
	return indexByObject
}

// ListEffectiveFieldPermissionsForUser returns resolved field permissions, not
// aggregated repository rows. All scopes project the same field evaluator.
func (ps *PermissionService) ListEffectiveFieldPermissionsForUser(ctx context.Context, userID, profileID string) ([]models.FieldPermissions, error) {
	fields, err := ps.projectEffectiveFieldAccessForUser(ctx, userID, profileID, nil)
	return fieldPermissionRows(fields), err
}

func (ps *PermissionService) ListEffectiveFieldPermissionsForUserObject(ctx context.Context, userID, profileID, objectAPIName string) ([]models.FieldPermissions, error) {
	if strings.TrimSpace(objectAPIName) == "" {
		return nil, errBoundedFieldPermissionProjectionUnavailable
	}
	fields, err := ps.projectEffectiveFieldAccessForUser(ctx, userID, profileID, []string{objectAPIName})
	return fieldPermissionRows(fields), err
}

func (ps *PermissionService) ListEffectiveFieldPermissionsForUserObjects(ctx context.Context, userID, profileID string, resources []string) ([]models.FieldPermissions, error) {
	if len(resources) == 0 {
		return nil, errBoundedFieldPermissionProjectionUnavailable
	}
	fields, err := ps.projectEffectiveFieldAccessForUser(ctx, userID, profileID, resources)
	return fieldPermissionRows(fields), err
}

func fieldPermissionRows(fields []contract.FieldAccessProjection) []models.FieldPermissions {
	if fields == nil {
		return nil
	}
	rows := make([]models.FieldPermissions, 0, len(fields))
	for _, field := range fields {
		rows = append(rows, models.FieldPermissions{
			SobjectType: field.ResourceType, Field: field.FieldName,
			PermissionsRead: field.CanRead, PermissionsEdit: field.CanEdit,
		})
	}
	return rows
}

type effectiveFieldPermissionBatchRepository interface {
	ListEffectiveFieldPermissionsForUserObjects(context.Context, string, string, []string) ([]models.FieldPermissions, error)
}

func normalizeFieldPermissionResources(resources []string) []string {
	normalized := make([]string, 0, len(resources))
	seen := make(map[string]bool, len(resources))
	for _, resource := range resources {
		resource = strings.TrimSpace(resource)
		key := strings.ToLower(resource)
		if key != "" && !seen[key] {
			normalized = append(normalized, resource)
			seen[key] = true
		}
	}
	sort.Strings(normalized)
	return normalized
}

// loadFieldPermissionInputs is the sole bulk loader of explicit FLS input rows.
// The request cache must use this input loader, never a resolved projection:
// missing-row fallback and create-only access depend on preserving row absence.
// nil resources is the explicit all-object administrative scope. A bounded
// request never falls back to it.
func (ps *PermissionService) loadFieldPermissionInputs(ctx context.Context, userID, profileID string, resources []string) ([]models.FieldPermissions, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	release, err := ps.acquireBoundedFieldPermissionProjection(ctx)
	if err != nil {
		return nil, err
	}
	defer release()
	return ps.readFieldPermissionInputs(ctx, userID, profileID, resources)
}

func (ps *PermissionService) readFieldPermissionInputs(ctx context.Context, userID, profileID string, resources []string) ([]models.FieldPermissions, error) {
	if ps == nil || ps.repo == nil {
		return nil, errBoundedFieldPermissionProjectionUnavailable
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if resources != nil {
		resources = normalizeFieldPermissionResources(resources)
		if len(resources) == 0 {
			return nil, errBoundedFieldPermissionProjectionUnavailable
		}
	}
	var err error
	start := time.Now()
	var rows []models.FieldPermissions
	defer func() { recordFieldPermissionBulkLoad(ctx, len(rows), time.Since(start)) }()
	switch {
	case resources == nil:
		rows, err = ps.repo.ListEffectiveFieldPermissionsForUser(ctx, userID, profileID)
	case len(resources) == 1:
		rows, err = ps.repo.ListEffectiveFieldPermissionsForUserObject(ctx, userID, profileID, resources[0])
	default:
		if batch, ok := ps.repo.(effectiveFieldPermissionBatchRepository); ok {
			rows, err = batch.ListEffectiveFieldPermissionsForUserObjects(ctx, userID, profileID, resources)
		} else {
			for _, resource := range resources {
				var selected []models.FieldPermissions
				selected, err = ps.repo.ListEffectiveFieldPermissionsForUserObject(ctx, userID, profileID, resource)
				if err != nil {
					break
				}
				rows = append(rows, selected...)
			}
		}
	}
	if err != nil {
		return nil, err
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if resources == nil {
		return rows, nil
	}
	selected := make(map[string]bool, len(resources))
	for _, resource := range resources {
		selected[strings.ToLower(resource)] = true
	}
	bounded := make([]models.FieldPermissions, 0, len(rows))
	for _, row := range rows {
		if selected[strings.ToLower(strings.TrimSpace(row.SobjectType))] {
			bounded = append(bounded, row)
		}
	}
	return bounded, nil
}

// projectEffectiveFieldAccessForUser resolves every admitted metadata field
// through the same evaluator as record reads and writes. Projection maps are
// temporary indexes over explicit input rows, not stored authorization facts.
func (ps *PermissionService) projectEffectiveFieldAccessForUser(ctx context.Context, userID, profileID string, resources []string) ([]contract.FieldAccessProjection, error) {
	if ps == nil || ps.metadata == nil {
		return nil, errBoundedFieldPermissionProjectionUnavailable
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	ctx = WithAccessInputCache(ctx)
	var schemas []*models.ObjectMetadata
	if resources == nil {
		schemas = ps.metadata.GetSchemas(ctx)
	} else {
		resources = normalizeFieldPermissionResources(resources)
		if len(resources) == 0 {
			return nil, errBoundedFieldPermissionProjectionUnavailable
		}
		for _, resource := range resources {
			if schema := ps.metadata.GetSchema(ctx, resource); schema != nil {
				schemas = append(schemas, schema)
			}
		}
	}
	user := &models.UserSession{ID: userID, ProfileID: profileID}
	super := ps.HasSuperUserCapability(ctx, user)
	// Keep the input load and CPU projection within the same bounded admission.
	if !super && len(schemas) > 0 {
		release, err := ps.acquireBoundedFieldPermissionProjection(ctx)
		if err != nil {
			return nil, err
		}
		defer release()
	}
	needsInputs := false
	for _, schema := range schemas {
		if schema != nil && !schema.UsesDerivedAccessPolicy() && len(schema.Fields) > 0 {
			needsInputs = true
		}
	}
	inputs := make(map[string]*models.FieldPermissions)
	if !super && needsInputs {
		rows, err := ps.readFieldPermissionInputs(ctx, userID, profileID, resources)
		if err != nil {
			return nil, err
		}
		for _, row := range rows {
			key := effectiveFieldPermissionCacheKey(row.SobjectType, row.Field)
			if key == "" {
				return nil, errBoundedFieldPermissionProjectionUnavailable
			}
			if previous := inputs[key]; previous != nil &&
				(previous.PermissionsRead != row.PermissionsRead || previous.PermissionsEdit != row.PermissionsEdit) {
				return nil, errBoundedFieldPermissionProjectionUnavailable
			}
			copy := row
			inputs[key] = &copy
		}
	}
	var projected []contract.FieldAccessProjection
	for _, schema := range schemas {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		if schema == nil {
			continue
		}
		evaluation := ps.newFieldAccessEvaluation(ctx, schema.APIName, schema, nil, user)
		evaluation.super = &super
		evaluation.loadPermission = func(_ context.Context, field string) (*models.FieldPermissions, error) {
			return inputs[effectiveFieldPermissionCacheKey(schema.APIName, field)], nil
		}
		for i := range schema.Fields {
			field := &schema.Fields[i]
			readable, err := evaluation.resolve(ctx, field.APIName, field, fieldAccessRead)
			if err != nil {
				return nil, err
			}
			editable, err := evaluation.resolve(ctx, field.APIName, field, fieldAccessEdit)
			if err != nil {
				return nil, err
			}
			// Permission views deliberately omit mutation-only intrinsic field
			// capabilities. Resolve insert authorization here; the caller also
			// applies the field's intrinsic createability from its own admitted
			// metadata view, just as it does for updateability.
			createable := false
			if evaluation.objectAllowed(ctx, constants.PermCreate) {
				createable, err = evaluation.resolve(ctx, field.APIName, field, fieldAccessCreate)
				if err != nil {
					return nil, err
				}
			}
			// Keep explicit denial results. Dropping them would turn false back into
			// row absence for record-view consumers and incorrectly activate fallback.
			projected = append(projected, contract.FieldAccessProjection{
				ResourceType: schema.APIName, FieldName: field.APIName,
				CanRead: readable, CanEdit: editable, CanCreate: createable,
			})
		}
	}
	sort.Slice(projected, func(i, j int) bool {
		left := strings.ToLower(projected[i].ResourceType) + "\x00" + strings.ToLower(projected[i].FieldName)
		right := strings.ToLower(projected[j].ResourceType) + "\x00" + strings.ToLower(projected[j].FieldName)
		return left < right
	})
	return projected, nil
}