package permissions

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/fieldpermissions"
	"github.com/nexuscrm/shared/pkg/models"
)

// countingShareRepo embeds saveTestRepo (full PermissionRepository impl returning
// zero values) and counts the access-input lookups relevant to issue #1585.
// All grants are empty so every referenced record is denied — the point of the
// test is the QUERY COUNT, not the decision.
type countingShareRepo struct {
	*saveTestRepo

	perRecordManualShare  int // GetManualShareAccessLevels (per (object,record,user))
	perRecordTeam         int // GetTeamMemberAccessLevel (per (object,record,user))
	perRecordGroup        int // IsUserInGroup (per (group,member))
	objectWideManualShare int // GetManualShareRecordIDs (per (object,user))
	objectWideTeam        int // GetTeamMemberRecordIDs (per (object,user))
	scopedManualShare     int // GetManualShareAccessLevelsByRecordIDs
	scopedTeam            int // GetTeamMemberAccessLevelsByRecordIDs
	batchGroups           int // GetUserGroupIDs (per user)
	sysPermLoads          int // GetEffectiveSystemPermissions (per (user,profile))
	customPermLoads       int // GetEffectiveCustomPermissions (per (user,profile))
	objPermLoads          int // LoadEffectiveObjectPermission (per (user,profile,object))
	fieldPermLoads        int // LoadEffectiveFieldPermission (per (user,profile,object,field))
	fieldPermBulkLoads    int // ListEffectiveFieldPermissionsForUser (per (user,profile))
	fieldPermObjectLoads  int // ListEffectiveFieldPermissionsForUserObject (per (user,profile,object))

	manualShareLevelsByRecord map[string][]string
	teamMemberAccessByRecord  map[string]string
	objectPermByName          map[string]*models.ObjectPermissions
	fieldPermByRef            map[string]*models.FieldPermissions
	userGroupIDs              []string
}

type concurrentObjectPermissionRepo struct {
	*saveTestRepo
	entered chan struct{}
	release chan struct{}
	calls   atomic.Int32
	active  atomic.Int32
	max     atomic.Int32
}

func (r *concurrentObjectPermissionRepo) LoadEffectiveObjectPermission(_ context.Context, _ *models.UserSession, objectAPIName string) (*models.ObjectPermissions, error) {
	r.calls.Add(1)
	active := r.active.Add(1)
	defer r.active.Add(-1)
	for {
		seen := r.max.Load()
		if active <= seen || r.max.CompareAndSwap(seen, active) {
			break
		}
	}
	r.entered <- struct{}{}
	<-r.release
	return &models.ObjectPermissions{SobjectType: objectAPIName, PermissionsRead: true}, nil
}

func (r *countingShareRepo) LoadEffectiveObjectPermission(ctx context.Context, user *models.UserSession, objectAPIName string) (*models.ObjectPermissions, error) {
	r.objPermLoads++
	if p, ok := r.objectPermByName[objectAPIName]; ok {
		return p, nil
	}
	return nil, nil
}

func (r *countingShareRepo) LoadEffectiveFieldPermission(ctx context.Context, user *models.UserSession, objectAPIName, fieldAPIName string) (*models.FieldPermissions, error) {
	r.fieldPermLoads++
	for raw, perm := range r.fieldPermByRef {
		if fieldpermissions.MatchesFieldReference(objectAPIName, raw, fieldAPIName) {
			cp := *perm
			return &cp, nil
		}
	}
	return nil, nil
}

func (r *countingShareRepo) ListEffectiveFieldPermissionsForUser(ctx context.Context, userID, profileID string) ([]models.FieldPermissions, error) {
	r.fieldPermBulkLoads++
	out := make([]models.FieldPermissions, 0, len(r.fieldPermByRef))
	for _, perm := range r.fieldPermByRef {
		if perm == nil {
			continue
		}
		out = append(out, *perm)
	}
	return out, nil
}

func (r *countingShareRepo) ListEffectiveFieldPermissionsForUserObject(_ context.Context, _, _, objectAPIName string) ([]models.FieldPermissions, error) {
	r.fieldPermObjectLoads++
	out := make([]models.FieldPermissions, 0, len(r.fieldPermByRef))
	for raw, perm := range r.fieldPermByRef {
		if perm == nil {
			continue
		}
		objectName, _ := fieldpermissions.SplitFieldReference(objectAPIName, raw)
		if strings.EqualFold(objectName, objectAPIName) {
			out = append(out, *perm)
		}
	}
	return out, nil
}

func (r *countingShareRepo) GetEffectiveSystemPermissions(ctx context.Context, userID, profileID string) (map[string]bool, error) {
	r.sysPermLoads++
	return map[string]bool{}, nil
}

func (r *countingShareRepo) GetEffectiveCustomPermissions(ctx context.Context, userID, profileID string) (map[string]bool, error) {
	r.customPermLoads++
	return map[string]bool{"ConvertLeads": true}, nil
}

func (r *countingShareRepo) GetManualShareAccessLevels(ctx context.Context, o, rec, u string) ([]string, error) {
	r.perRecordManualShare++
	return nil, nil
}
func (r *countingShareRepo) GetTeamMemberAccessLevel(ctx context.Context, o, rec, u string) (*string, error) {
	r.perRecordTeam++
	return nil, nil
}
func (r *countingShareRepo) IsUserInGroup(ctx context.Context, g, m string) (bool, error) {
	r.perRecordGroup++
	return false, nil
}
func (r *countingShareRepo) GetManualShareRecordIDs(ctx context.Context, o, u string) ([]string, error) {
	r.objectWideManualShare++
	return nil, nil
}
func (r *countingShareRepo) GetGroupMemberIDs(context.Context, []string) ([]string, error) {
	return nil, nil
}
func (r *countingShareRepo) GetTeamMemberRecordIDs(ctx context.Context, o, u string) ([]string, error) {
	r.objectWideTeam++
	return nil, nil
}
func (r *countingShareRepo) GetUserGroupIDs(ctx context.Context, u string) ([]string, error) {
	r.batchGroups++
	return append([]string(nil), r.userGroupIDs...), nil
}
func (r *countingShareRepo) GetManualShareAccessLevelsByRecordIDs(ctx context.Context, o, u string, recordIDs []string) (map[string][]string, error) {
	r.scopedManualShare++
	out := make(map[string][]string)
	for _, id := range recordIDs {
		if levels := r.manualShareLevelsByRecord[id]; len(levels) > 0 {
			out[id] = append([]string(nil), levels...)
		}
	}
	return out, nil
}
func (r *countingShareRepo) GetTeamMemberAccessLevelsByRecordIDs(ctx context.Context, o, u string, recordIDs []string) (map[string]string, error) {
	r.scopedTeam++
	out := make(map[string]string)
	for _, id := range recordIDs {
		if level := r.teamMemberAccessByRecord[id]; level != "" {
			out[id] = level
		}
	}
	return out, nil
}

// cacheTestMetadata is a minimal metadataProvider: one Private object, no rules.
type cacheTestMetadata struct {
	schema *models.ObjectMetadata
	rules  []*models.SharingRule
}

type parentAggregateMetadata struct {
	schemas       map[string]*models.ObjectMetadata
	rules         map[string][]*models.SharingRule
	sharingParent map[string]*models.Relationship
}

func (m *parentAggregateMetadata) GetSchema(_ context.Context, apiName string) *models.ObjectMetadata {
	return m.schemas[apiName]
}

func (m *parentAggregateMetadata) GetSchemas(context.Context) []*models.ObjectMetadata {
	out := make([]*models.ObjectMetadata, 0, len(m.schemas))
	for _, schema := range m.schemas {
		out = append(out, schema)
	}
	return out
}

func (m *parentAggregateMetadata) GetRelationships(context.Context, string) []*models.Relationship {
	return nil
}

func (m *parentAggregateMetadata) GetSharingParentRelationship(_ context.Context, childAPIName string) *models.Relationship {
	return m.sharingParent[childAPIName]
}

func (m *parentAggregateMetadata) GetParentControlledObject(_ context.Context, schema *models.ObjectMetadata) string {
	if schema == nil || schema.SharingModel != constants.SharingModelControlledByParent {
		return ""
	}
	if rel := m.sharingParent[schema.APIName]; rel != nil {
		return rel.ParentObjectAPIName
	}
	return ""
}

func (m *parentAggregateMetadata) GetSharingRules(_ context.Context, objectAPIName string) []*models.SharingRule {
	return m.rules[objectAPIName]
}

func (m *parentAggregateMetadata) GetRecordTypes(context.Context, string) ([]*models.RecordType, error) {
	return nil, nil
}

type parentAggregateRepo struct {
	*countingShareRepo
	records      map[string]models.SObject
	getByIDCalls int
}

func (r *parentAggregateRepo) GetRecordByID(_ context.Context, objectAPIName, recordID string) (models.SObject, error) {
	r.getByIDCalls++
	return r.records[objectAPIName+":"+recordID], nil
}

func (m *cacheTestMetadata) GetSchema(ctx context.Context, apiName string) *models.ObjectMetadata {
	return m.schema
}
func (m *cacheTestMetadata) GetSchemas(ctx context.Context) []*models.ObjectMetadata {
	return []*models.ObjectMetadata{m.schema}
}
func (m *cacheTestMetadata) GetRelationships(ctx context.Context, o string) []*models.Relationship {
	return nil
}
func (m *cacheTestMetadata) GetSharingParentRelationship(ctx context.Context, c string) *models.Relationship {
	return nil
}
func (m *cacheTestMetadata) GetParentControlledObject(ctx context.Context, s *models.ObjectMetadata) string {
	return ""
}
func (m *cacheTestMetadata) GetSharingRules(ctx context.Context, o string) []*models.SharingRule {
	return m.rules
}

func (m *cacheTestMetadata) GetRecordTypes(context.Context, string) ([]*models.RecordType, error) {
	return nil, nil
}

func newCacheTestService(repo *countingShareRepo) *PermissionService {
	md := &cacheTestMetadata{schema: &models.ObjectMetadata{
		APIName:      "RefObj",
		SharingModel: constants.SharingModelPrivate,
	}}
	return NewPermissionService(repo, md, nil)
}

func makeRefRecords(n int) []models.SObject {
	recs := make([]models.SObject, n)
	for i := 0; i < n; i++ {
		recs[i] = models.SObject{
			constants.FieldID:      "rec-" + string(rune('A'+i%26)) + string(rune('0'+i/26)),
			constants.FieldOwnerID: "owner-other", // not the requesting user → Private denies
		}
	}
	return recs
}

// TestAccessInputCache_BatchesPerRecordRLS proves the request-scoped cache turns
// O(N) per-record access-input queries into O(1) scoped batch queries for the
// current candidate record IDs, while producing the same (deny) decision.
func TestAccessInputCache_BatchesPerRecordRLS(t *testing.T) {
	const n = 50
	repo := &countingShareRepo{saveTestRepo: &saveTestRepo{}}
	svc := newCacheTestService(repo)
	schema := &models.ObjectMetadata{APIName: "RefObj", SharingModel: constants.SharingModelPrivate}
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	records := makeRefRecords(n)

	ctx := WithAccessInputCache(context.Background())
	svc.PrimeReadAccessInputCache(ctx, schema.APIName, records, user)
	for _, rec := range records {
		if svc.CheckRecordAccess(ctx, schema, rec, constants.PermRead, user) {
			t.Fatalf("expected deny for Private record not owned by user")
		}
	}

	if repo.batchGroups != 0 {
		t.Errorf("GetUserGroupIDs must not be used on record-read checks, got %d", repo.batchGroups)
	}
	if repo.scopedManualShare != 1 {
		t.Errorf("GetManualShareAccessLevelsByRecordIDs: want 1 scoped call, got %d", repo.scopedManualShare)
	}
	if repo.scopedTeam != 1 {
		t.Errorf("GetTeamMemberAccessLevelsByRecordIDs: want 1 scoped call, got %d", repo.scopedTeam)
	}
	if repo.objectWideManualShare != 0 || repo.objectWideTeam != 0 {
		t.Errorf("object-wide share/team loaders must not run; got manualShare=%d team=%d",
			repo.objectWideManualShare, repo.objectWideTeam)
	}
	if repo.perRecordGroup != 1 || repo.perRecordManualShare != 0 || repo.perRecordTeam != 0 {
		t.Errorf("expected group membership point-memoized once and 0 per-record share/team queries; got group=%d manualShare=%d team=%d",
			repo.perRecordGroup, repo.perRecordManualShare, repo.perRecordTeam)
	}
	// System permissions are probed twice per record-access check (ModifyAllData
	// + ViewAllData); the cache collapses all N*2 probes to a single load.
	if repo.sysPermLoads != 1 {
		t.Errorf("GetEffectiveSystemPermissions: want 1 load with cache, got %d", repo.sysPermLoads)
	}
}

func TestAccessInputCache_CachesCustomPermissionsForHasPermission(t *testing.T) {
	repo := &countingShareRepo{saveTestRepo: &saveTestRepo{}}
	svc := newCacheTestService(repo)
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	ctx := WithAccessInputCache(context.Background())

	for i := 0; i < 5; i++ {
		if !svc.HasPermission(ctx, user, "ConvertLeads") {
			t.Fatalf("HasPermission(ConvertLeads) = false; want true")
		}
	}

	if repo.sysPermLoads != 1 {
		t.Fatalf("GetEffectiveSystemPermissions loads = %d; want 1", repo.sysPermLoads)
	}
	if repo.customPermLoads != 1 {
		t.Fatalf("GetEffectiveCustomPermissions loads = %d; want 1", repo.customPermLoads)
	}
}

func TestSystemPermissionsFromRequestContextAvoidsRepositoryLoad(t *testing.T) {
	repo := &countingShareRepo{saveTestRepo: &saveTestRepo{}}
	svc := newCacheTestService(repo)
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	ctx := WithEffectivePermissions(
		WithAccessInputCache(context.Background()),
		user.ID,
		user.ProfileID,
		map[string]bool{string(constants.PermModifyAllData): true},
		nil,
	)

	for i := 0; i < 5; i++ {
		if !svc.CheckSystemPermission(ctx, user, string(constants.PermModifyAllData)) {
			t.Fatalf("CheckSystemPermission(ModifyAllData) = false; want true")
		}
	}

	if repo.sysPermLoads != 0 {
		t.Fatalf("GetEffectiveSystemPermissions loads = %d; want 0 when request context already has system permissions", repo.sysPermLoads)
	}
}

func TestCustomPermissionsFromRequestContextAvoidsRepositoryLoad(t *testing.T) {
	repo := &countingShareRepo{saveTestRepo: &saveTestRepo{}}
	svc := newCacheTestService(repo)
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	ctx := WithEffectivePermissions(
		WithAccessInputCache(context.Background()),
		user.ID,
		user.ProfileID,
		map[string]bool{},
		map[string]bool{"ConvertLeads": true},
	)

	for i := 0; i < 5; i++ {
		if !svc.HasPermission(ctx, user, "ConvertLeads") {
			t.Fatalf("HasPermission(ConvertLeads) = false; want true")
		}
	}

	if repo.sysPermLoads != 0 {
		t.Fatalf("GetEffectiveSystemPermissions loads = %d; want 0 when request context already has system permissions", repo.sysPermLoads)
	}
	if repo.customPermLoads != 0 {
		t.Fatalf("GetEffectiveCustomPermissions loads = %d; want 0 when request context already has custom permissions", repo.customPermLoads)
	}
}

func TestEffectiveRecordReadProjectionAvoidsRepeatedPermissionRepositoryLoads(t *testing.T) {
	repo := &countingShareRepo{saveTestRepo: &saveTestRepo{}}
	svc := newCacheTestService(repo)
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	ctx, err := WithEffectiveRecordReadProjection(
		context.Background(), user.ID, user.ProfileID, "RefObj",
		map[string]bool{string(constants.PermViewAllData): false}, nil,
		models.ObjectPermissions{SobjectType: "RefObj", PermissionsRead: true},
		[]models.FieldPermissions{{SobjectType: "RefObj", Field: "Name", PermissionsRead: true}},
	)
	if err != nil {
		t.Fatal(err)
	}
	if svc.CheckSystemPermission(ctx, user, string(constants.PermViewAllData)) {
		t.Fatal("explicit false system permission became granted")
	}
	objectPermission, err := svc.loadEffectiveObjectPermission(ctx, user, "RefObj")
	if err != nil || objectPermission == nil || !objectPermission.PermissionsRead {
		t.Fatalf("object permission = %#v, %v", objectPermission, err)
	}
	fieldPermission, err := svc.loadEffectiveFieldPermission(ctx, user, "RefObj", "Name")
	if err != nil || fieldPermission == nil || !fieldPermission.PermissionsRead {
		t.Fatalf("field permission = %#v, %v", fieldPermission, err)
	}
	missingPermission, err := svc.loadEffectiveFieldPermission(ctx, user, "RefObj", "Secret")
	if err != nil || missingPermission != nil {
		t.Fatalf("missing field permission = %#v, %v", missingPermission, err)
	}
	ctx, err = WithEffectiveRecordReadProjection(
		ctx, user.ID, user.ProfileID, "ParentObj",
		map[string]bool{string(constants.PermViewAllData): false}, nil,
		models.ObjectPermissions{SobjectType: "ParentObj", PermissionsRead: true, PermissionsEdit: true},
		[]models.FieldPermissions{{SobjectType: "ParentObj", Field: "Name", PermissionsRead: true}},
	)
	if err != nil {
		t.Fatal(err)
	}
	parentPermission, err := svc.loadEffectiveObjectPermission(ctx, user, "ParentObj")
	if err != nil || parentPermission == nil || !parentPermission.PermissionsEdit {
		t.Fatalf("parent permission = %#v, %v", parentPermission, err)
	}
	parentFieldPermission, err := svc.loadEffectiveFieldPermission(ctx, user, "ParentObj", "Name")
	if err != nil || parentFieldPermission == nil || !parentFieldPermission.PermissionsRead {
		t.Fatalf("parent field permission = %#v, %v", parentFieldPermission, err)
	}
	if repo.sysPermLoads != 0 || repo.objPermLoads != 0 || repo.fieldPermObjectLoads != 0 || repo.fieldPermLoads != 0 {
		t.Fatalf("projection fell back to repository: system=%d object=%d field-object=%d field=%d",
			repo.sysPermLoads, repo.objPermLoads, repo.fieldPermObjectLoads, repo.fieldPermLoads)
	}
}

func TestRequestContextPermissionsDoNotApplyToDifferentUser(t *testing.T) {
	repo := &countingShareRepo{saveTestRepo: &saveTestRepo{}}
	svc := newCacheTestService(repo)
	sessionUser := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	otherUser := &models.UserSession{ID: "user-2", ProfileID: "standard_user"}
	ctx := WithEffectivePermissions(
		WithAccessInputCache(context.Background()),
		sessionUser.ID,
		sessionUser.ProfileID,
		map[string]bool{string(constants.PermModifyAllData): true},
		nil,
	)

	if svc.CheckSystemPermission(ctx, otherUser, string(constants.PermModifyAllData)) {
		t.Fatalf("request context permissions for %s must not grant %s", sessionUser.ID, otherUser.ID)
	}
	if repo.sysPermLoads != 1 {
		t.Fatalf("GetEffectiveSystemPermissions loads = %d; want 1 fallback for different user", repo.sysPermLoads)
	}
}

// TestAccessInputCache_AbsentFallsBackPerRecord proves fail-closed fallback: with
// no cache installed, the existing per-record path runs unchanged (N queries).
func TestAccessInputCache_AbsentFallsBackPerRecord(t *testing.T) {
	const n = 20
	repo := &countingShareRepo{saveTestRepo: &saveTestRepo{}}
	svc := newCacheTestService(repo)
	schema := &models.ObjectMetadata{APIName: "RefObj", SharingModel: constants.SharingModelPrivate}
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	records := makeRefRecords(n)

	ctx := context.Background() // NO cache installed
	for _, rec := range records {
		_ = svc.CheckRecordAccess(ctx, schema, rec, constants.PermRead, user)
	}

	if repo.batchGroups != 0 || repo.objectWideManualShare != 0 || repo.objectWideTeam != 0 || repo.scopedManualShare != 0 || repo.scopedTeam != 0 {
		t.Errorf("expected 0 batch calls without cache; got groups=%d objectManual=%d objectTeam=%d scopedManual=%d scopedTeam=%d",
			repo.batchGroups, repo.objectWideManualShare, repo.objectWideTeam, repo.scopedManualShare, repo.scopedTeam)
	}
	if repo.perRecordManualShare != n || repo.perRecordTeam != n {
		t.Errorf("expected %d per-record manual-share/team calls without cache; got manualShare=%d team=%d",
			n, repo.perRecordManualShare, repo.perRecordTeam)
	}
	// Without the cache, system permissions are reloaded on every probe
	// (2 per record: ModifyAllData + ViewAllData).
	if repo.sysPermLoads != 2*n {
		t.Errorf("expected %d system-permission loads without cache, got %d", 2*n, repo.sysPermLoads)
	}
}

// TestAccessInputCache_UnprimedReadFallsBackPerRecord guards the hot-path
// regression from PR #1593: installing a request cache must not make a single
// record read load every shared/team-member record for an object. Without an
// explicit candidate-set prime, READ falls back to the original point lookups.
func TestAccessInputCache_UnprimedReadFallsBackPerRecord(t *testing.T) {
	const n = 10
	repo := &countingShareRepo{saveTestRepo: &saveTestRepo{}}
	svc := newCacheTestService(repo)
	schema := &models.ObjectMetadata{APIName: "RefObj", SharingModel: constants.SharingModelPrivate}
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	records := makeRefRecords(n)

	ctx := WithAccessInputCache(context.Background())
	for _, rec := range records {
		_ = svc.CheckRecordAccess(ctx, schema, rec, constants.PermRead, user)
	}

	if repo.objectWideManualShare != 0 || repo.objectWideTeam != 0 {
		t.Fatalf("unprimed read used object-wide loaders: manualShare=%d team=%d", repo.objectWideManualShare, repo.objectWideTeam)
	}
	if repo.scopedManualShare != 0 || repo.scopedTeam != 0 {
		t.Fatalf("unprimed read used scoped loaders: manualShare=%d team=%d", repo.scopedManualShare, repo.scopedTeam)
	}
	if repo.perRecordManualShare != n || repo.perRecordTeam != n {
		t.Fatalf("unprimed read should use %d point lookups; got manualShare=%d team=%d",
			n, repo.perRecordManualShare, repo.perRecordTeam)
	}
}

func TestAccessInputCache_ScopedReadGrantsRespectAccessLevel(t *testing.T) {
	repo := &countingShareRepo{
		saveTestRepo: &saveTestRepo{},
		manualShareLevelsByRecord: map[string][]string{
			"rec-A0": []string{"Nope"},
			"rec-B0": []string{constants.PermRead},
		},
		teamMemberAccessByRecord: map[string]string{
			"rec-C0": "Nope",
			"rec-D0": constants.PermEdit,
		},
	}
	svc := newCacheTestService(repo)
	schema := &models.ObjectMetadata{APIName: "RefObj", SharingModel: constants.SharingModelPrivate}
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	records := makeRefRecords(4)
	ctx := WithAccessInputCache(context.Background())
	svc.PrimeReadAccessInputCache(ctx, schema.APIName, records, user)

	if svc.CheckRecordAccess(ctx, schema, records[0], constants.PermRead, user) {
		t.Fatal("invalid manual share level must not grant read")
	}
	if !svc.CheckRecordAccess(ctx, schema, records[1], constants.PermRead, user) {
		t.Fatal("Read manual share level should grant read")
	}
	if svc.CheckRecordAccess(ctx, schema, records[2], constants.PermRead, user) {
		t.Fatal("invalid team access level must not grant read")
	}
	if !svc.CheckRecordAccess(ctx, schema, records[3], constants.PermRead, user) {
		t.Fatal("Edit team access level should grant read")
	}
	if repo.perRecordManualShare != 0 || repo.perRecordTeam != 0 {
		t.Fatalf("primed records should not use point share/team lookups; got manualShare=%d team=%d",
			repo.perRecordManualShare, repo.perRecordTeam)
	}
	if repo.objectWideManualShare != 0 || repo.objectWideTeam != 0 {
		t.Fatalf("scoped prime must not use object-wide loaders; got manualShare=%d team=%d",
			repo.objectWideManualShare, repo.objectWideTeam)
	}
}

func TestAccessInputCache_ScopedGrantLevelsServeReadEditDelete(t *testing.T) {
	repo := &countingShareRepo{
		saveTestRepo: &saveTestRepo{},
		manualShareLevelsByRecord: map[string][]string{
			"rec-A0": {constants.PermEdit},
			"rec-B0": {"Full"},
		},
		teamMemberAccessByRecord: map[string]string{
			"rec-C0": constants.PermEdit,
		},
	}
	svc := newCacheTestService(repo)
	schema := &models.ObjectMetadata{APIName: "RefObj", SharingModel: constants.SharingModelPrivate}
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	records := makeRefRecords(3)
	ctx := WithAccessInputCache(context.Background())
	svc.PrimeRecordAccessInputCache(ctx, schema.APIName, records, user)

	assertAccess := func(record models.SObject, operation string, want bool) {
		t.Helper()
		if got := svc.CheckRecordAccess(ctx, schema, record, operation, user); got != want {
			t.Fatalf("record=%v operation=%s access=%t, want %t", record[constants.FieldID], operation, got, want)
		}
	}
	assertAccess(records[0], constants.PermRead, true)
	assertAccess(records[0], constants.PermEdit, true)
	assertAccess(records[0], constants.PermDelete, false)
	assertAccess(records[1], constants.PermRead, true)
	assertAccess(records[1], constants.PermEdit, true)
	assertAccess(records[1], constants.PermDelete, true)
	assertAccess(records[2], constants.PermRead, true)
	assertAccess(records[2], constants.PermEdit, true)
	assertAccess(records[2], constants.PermDelete, false)

	if repo.scopedManualShare != 1 || repo.scopedTeam != 1 {
		t.Fatalf("scoped loads manual=%d team=%d, want 1/1", repo.scopedManualShare, repo.scopedTeam)
	}
	if repo.perRecordManualShare != 0 || repo.perRecordTeam != 0 {
		t.Fatalf("point loads manual=%d team=%d, want 0/0", repo.perRecordManualShare, repo.perRecordTeam)
	}
}

// TestAccessInputCache_GroupMembershipRouting proves group-membership questions
// stay point-memoized per (group, member), including for the session user, so a
// single-record read never expands into a full user-group preload.
func TestAccessInputCache_GroupMembershipRouting(t *testing.T) {
	repo := &countingShareRepo{saveTestRepo: &saveTestRepo{}}
	ps := &PermissionService{repo: repo}
	ctx := WithAccessInputCache(context.Background())

	// Session user: served from point memoization, not full GetUserGroupIDs.
	for i := 0; i < 4; i++ {
		_ = ps.isUserInGroupForSession(ctx, "group-1", "user-1", "user-1")
	}
	if repo.batchGroups != 0 || repo.perRecordGroup != 1 {
		t.Fatalf("session-user path: want batchGroups=0 perRecordGroup=1, got batchGroups=%d perRecordGroup=%d",
			repo.batchGroups, repo.perRecordGroup)
	}

	// Non-session member (e.g. record owner during sharing-rule source eval):
	// point-memoized, so a repeated (group, member) pair is one query.
	for i := 0; i < 4; i++ {
		_ = ps.isUserInGroupForSession(ctx, "group-1", "owner-9", "user-1")
	}
	if repo.perRecordGroup != 2 {
		t.Fatalf("non-session member: want one additional per-pair query for a repeated pair, got %d total", repo.perRecordGroup)
	}

	// A distinct (group, member) pair is a separate key → a separate query.
	_ = ps.cachedIsMemberOfGroup(ctx, "group-2", "owner-9")
	if repo.perRecordGroup != 3 {
		t.Fatalf("distinct pair should miss cache; want 3 total per-pair queries, got %d", repo.perRecordGroup)
	}
}

func TestAggregateRecordAccessBatchesSharingGroupMembership(t *testing.T) {
	groupRead, groupEdit := "group-read", "group-edit"
	schema := &models.ObjectMetadata{
		APIName: "Account", SharingModel: constants.SharingModelPrivate,
		Fields: []models.FieldMetadata{{APIName: constants.FieldID, Type: constants.FieldTypeString}},
	}
	repo := &countingShareRepo{
		saveTestRepo: &saveTestRepo{},
		objectPermByName: map[string]*models.ObjectPermissions{
			"Account": {SobjectType: "Account", PermissionsRead: true, PermissionsEdit: true, PermissionsDelete: true},
		},
		userGroupIDs: []string{groupEdit},
	}
	metadata := &cacheTestMetadata{schema: schema, rules: []*models.SharingRule{
		{ID: "read-rule", ObjectAPIName: "Account", Criteria: `[{"field":"Id","op":"!=","value":""}]`, AccessLevel: constants.PermRead, ShareWithGroupID: &groupRead},
		{ID: "edit-rule", ObjectAPIName: "Account", Criteria: `[{"field":"Id","op":"!=","value":""}]`, AccessLevel: constants.PermEdit, ShareWithGroupID: &groupEdit},
	}}
	svc := NewPermissionService(repo, metadata, nil)
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	record := models.SObject{constants.FieldID: "account-1", constants.FieldOwnerID: "owner-other"}
	ctx := WithAccessInputCache(context.Background())
	svc.PrimeRecordAccessInputCache(ctx, schema.APIName, []models.SObject{record}, user)

	access := svc.CheckRecordAccessAll(ctx, schema, record, user)
	if !access.HasReadAccess || !access.HasEditAccess || access.HasDeleteAccess {
		t.Fatalf("aggregate access = %#v, want read/edit and no delete", access)
	}
	if repo.batchGroups != 1 || repo.perRecordGroup != 0 {
		t.Fatalf("group membership loads batch=%d point=%d, want 1/0", repo.batchGroups, repo.perRecordGroup)
	}
}

func TestAggregateRecordAccessBatchesParentSharingGroupMembership(t *testing.T) {
	groupRead, groupEdit := "group-read", "group-edit"
	contact := &models.ObjectMetadata{
		APIName: "Contact", SharingModel: constants.SharingModelControlledByParent,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString},
			{APIName: "AccountId", Type: constants.FieldTypeReference, ReferenceTo: []string{"Account"}},
		},
	}
	account := &models.ObjectMetadata{
		APIName: "Account", SharingModel: constants.SharingModelPrivate,
		Fields: []models.FieldMetadata{{APIName: constants.FieldID, Type: constants.FieldTypeString}},
	}
	metadata := &parentAggregateMetadata{
		schemas: map[string]*models.ObjectMetadata{"Contact": contact, "Account": account},
		rules: map[string][]*models.SharingRule{
			"Account": {
				{ID: "read-rule", ObjectAPIName: "Account", Criteria: `[{"field":"Id","op":"!=","value":""}]`, AccessLevel: constants.PermRead, ShareWithGroupID: &groupRead},
				{ID: "edit-rule", ObjectAPIName: "Account", Criteria: `[{"field":"Id","op":"!=","value":""}]`, AccessLevel: constants.PermEdit, ShareWithGroupID: &groupEdit},
			},
		},
		sharingParent: map[string]*models.Relationship{
			"Contact": {
				ChildObjectAPIName:  "Contact",
				ParentObjectAPIName: "Account",
				FieldAPIName:        "AccountId",
				IsSharingParent:     true,
			},
		},
	}
	counting := &countingShareRepo{
		saveTestRepo: &saveTestRepo{},
		objectPermByName: map[string]*models.ObjectPermissions{
			"Contact": {SobjectType: "Contact", PermissionsRead: true, PermissionsEdit: true, PermissionsDelete: true},
			"Account": {SobjectType: "Account", PermissionsRead: true, PermissionsEdit: true, PermissionsDelete: true},
		},
		userGroupIDs: []string{groupEdit},
	}
	repo := &parentAggregateRepo{
		countingShareRepo: counting,
		records: map[string]models.SObject{
			"Account:account-1": {constants.FieldID: "account-1", constants.FieldOwnerID: "owner-other"},
		},
	}
	svc := NewPermissionService(repo, metadata, nil)
	user := &models.UserSession{ID: "user-1", ProfileID: "standard-user"}
	record := models.SObject{
		constants.FieldID:      "contact-1",
		constants.FieldOwnerID: "owner-other",
		"AccountId":            "account-1",
	}

	access := svc.CheckRecordAccessAll(context.Background(), contact, record, user)
	if !access.HasReadAccess || !access.HasEditAccess {
		t.Fatalf("aggregate parent access = %#v, want read/edit", access)
	}
	if repo.batchGroups != 1 || repo.perRecordGroup != 0 {
		t.Fatalf("parent group membership loads batch=%d point=%d, want 1/0", repo.batchGroups, repo.perRecordGroup)
	}
	if repo.getByIDCalls != 1 {
		t.Fatalf("parent record loads = %d, want exactly one shared by prefetch and Read/Edit/Delete", repo.getByIDCalls)
	}
}

// TestAccessInputCache_PointMemoizationFallsBack proves fail-closed fallback for
// the point path: with no cache installed, every call queries the repository.
func TestAccessInputCache_PointMemoizationFallsBack(t *testing.T) {
	repo := &countingShareRepo{saveTestRepo: &saveTestRepo{}}
	ps := &PermissionService{repo: repo}

	ctx := context.Background() // NO cache installed
	for i := 0; i < 3; i++ {
		_ = ps.cachedIsMemberOfGroup(ctx, "group-1", "owner-1")
	}
	if repo.perRecordGroup != 3 {
		t.Fatalf("without cache each membership check must query; want 3, got %d", repo.perRecordGroup)
	}
}

// TestAccessInputCache_MemoizesOwnerRoleID proves getRecordOwnerRoleID resolves
// each distinct owner's role at most once per request — including owners with no
// role (nil), so a list page of records owned by the same user issues one lookup.
func TestAccessInputCache_MemoizesOwnerRoleID(t *testing.T) {
	roleID := "role-1"
	userRepo := &countingUserRoleRepo{roleIDs: map[string]*string{"owner-1": &roleID}}
	ps := &PermissionService{userRepo: userRepo}
	ctx := WithAccessInputCache(context.Background())

	first := ps.getRecordOwnerRoleID(ctx, "owner-1")
	second := ps.getRecordOwnerRoleID(ctx, "owner-1")
	if first == nil || second == nil || *first != roleID || *second != roleID {
		t.Fatalf("expected memoized role %q twice, got first=%v second=%v", roleID, first, second)
	}
	if userRepo.getUserRoleCalls != 1 {
		t.Fatalf("expected 1 lookup for a repeated owner, got %d", userRepo.getUserRoleCalls)
	}

	// A different owner (with no role) is a distinct key; its nil result is also
	// memoized, so a repeat does not re-query.
	_ = ps.getRecordOwnerRoleID(ctx, "owner-2")
	_ = ps.getRecordOwnerRoleID(ctx, "owner-2")
	if userRepo.getUserRoleCalls != 2 {
		t.Fatalf("expected 2 lookups (owner-1 + owner-2 once each), got %d", userRepo.getUserRoleCalls)
	}
}

// countingUserRoleRepo is a minimal ports.UserRepository that counts GetUserRoleID
// calls; all other methods are no-ops. It exists to assert owner-role memoization.
type countingUserRoleRepo struct {
	getUserRoleCalls int
	roleIDs          map[string]*string
}

func (r *countingUserRoleRepo) GetUserRoleID(_ context.Context, userID string) (*string, error) {
	r.getUserRoleCalls++
	return r.roleIDs[userID], nil
}
func (r *countingUserRoleRepo) CheckUserExistsByEmail(context.Context, string) (bool, error) {
	return false, nil
}
func (r *countingUserRoleRepo) CheckUserExistsByID(context.Context, string) (bool, error) {
	return false, nil
}
func (r *countingUserRoleRepo) CheckEmailConflict(context.Context, string, string) (bool, error) {
	return false, nil
}
func (r *countingUserRoleRepo) UpdateUser(context.Context, string, map[string]interface{}) error {
	return nil
}
func (r *countingUserRoleRepo) DeleteUser(context.Context, string) error { return nil }
func (r *countingUserRoleRepo) FindAll(context.Context) ([]*models.User, error) {
	return nil, nil
}
func (r *countingUserRoleRepo) FindUserByEmailWithPassword(context.Context, string) (*ports.UserWithPassword, error) {
	return nil, nil
}
func (r *countingUserRoleRepo) FindUserByIDWithPassword(context.Context, string) (*ports.UserWithPassword, error) {
	return nil, nil
}
func (r *countingUserRoleRepo) UpdatePassword(context.Context, string, string) error { return nil }
func (r *countingUserRoleRepo) GetUserByID(context.Context, string) (*models.User, error) {
	return nil, nil
}
func (r *countingUserRoleRepo) CreateUser(context.Context, *models.User) error { return nil }
func (r *countingUserRoleRepo) GetFirstUserIDByProfile(context.Context, string) (string, error) {
	return "", nil
}
func (r *countingUserRoleRepo) GetRoleNameByID(context.Context, string) (string, error) {
	return "", nil
}
func (r *countingUserRoleRepo) GetProfileNameByID(context.Context, string) (string, error) {
	return "", nil
}
func (r *countingUserRoleRepo) GetUserIDsByRoleIDs(context.Context, []string) ([]string, error) {
	return nil, nil
}
func (r *countingUserRoleRepo) ResetFailedLoginAttempts(context.Context, string) error { return nil }
func (r *countingUserRoleRepo) IncrementFailedAndLockIfNeeded(context.Context, string, int, time.Duration) (bool, error) {
	return false, nil
}

var _ ports.UserRepository = (*countingUserRoleRepo)(nil)

func TestAccessInputCache_MemoizesObjectPermission(t *testing.T) {
	grant := &models.ObjectPermissions{
		SobjectType:       "Lead",
		PermissionsRead:   true,
		PermissionsCreate: true,
		PermissionsEdit:   true,
	}
	repo := &countingShareRepo{
		saveTestRepo:     &saveTestRepo{},
		objectPermByName: map[string]*models.ObjectPermissions{"Lead": grant},
	}
	svc := newCacheTestService(repo)
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	ctx := WithAccessInputCache(context.Background())

	for i := 0; i < 200; i++ {
		if !svc.loadAndCheck(ctx, user, "Lead", constants.PermRead) {
			t.Fatal("expected Read grant for Lead")
		}
		_ = svc.loadAndCheck(ctx, user, "Lead", constants.PermEdit)
	}

	if repo.objPermLoads != 1 {
		t.Fatalf("LoadEffectiveObjectPermission: want 1 load with memo, got %d", repo.objPermLoads)
	}
}

func (ps *PermissionService) loadAndCheck(ctx context.Context, user *models.UserSession, obj, op string) bool {
	perm, err := ps.loadEffectiveObjectPermission(ctx, user, obj)
	if err != nil || perm == nil {
		return false
	}
	switch op {
	case constants.PermRead:
		return perm.PermissionsRead
	case constants.PermCreate:
		return perm.PermissionsCreate
	case constants.PermEdit:
		return perm.PermissionsEdit
	case constants.PermDelete:
		return perm.PermissionsDelete
	}
	return false
}

func TestAccessInputCache_ObjectPermissionValueCorrect(t *testing.T) {
	grant := &models.ObjectPermissions{
		SobjectType:                 "Lead",
		PermissionsRead:             true,
		PermissionsEdit:             true,
		PermissionsViewAllRecords:   true,
		PermissionsModifyAllRecords: false,
		PermissionsCreate:           false,
		PermissionsDelete:           false,
	}
	repo := &countingShareRepo{
		saveTestRepo:     &saveTestRepo{},
		objectPermByName: map[string]*models.ObjectPermissions{"Lead": grant},
	}
	svc := newCacheTestService(repo)
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	ctx := WithAccessInputCache(context.Background())

	got, err := svc.loadEffectiveObjectPermission(ctx, user, "Lead")
	if err != nil || got == nil {
		t.Fatalf("loadEffectiveObjectPermission(Lead) = (%v, %v); want grant", got, err)
	}
	if *got != *grant {
		t.Fatalf("memo grant mismatch: got %+v want %+v", *got, *grant)
	}

	for i := 0; i < 3; i++ {
		none, err := svc.loadEffectiveObjectPermission(ctx, user, "Ghost")
		if err != nil || none != nil {
			t.Fatalf("loadEffectiveObjectPermission(Ghost) = (%v, %v); want (nil, nil)", none, err)
		}
	}
	if repo.objPermLoads != 2 {
		t.Fatalf("LoadEffectiveObjectPermission loads = %d; want 2 (Lead + memoized-nil Ghost)", repo.objPermLoads)
	}
}

func TestAccessInputCache_ObjectPermissionAbsentFallsBack(t *testing.T) {
	grant := &models.ObjectPermissions{SobjectType: "Lead", PermissionsRead: true}
	repo := &countingShareRepo{
		saveTestRepo:     &saveTestRepo{},
		objectPermByName: map[string]*models.ObjectPermissions{"Lead": grant},
	}
	svc := newCacheTestService(repo)
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}

	ctx := context.Background()
	for i := 0; i < 3; i++ {
		_, _ = svc.loadEffectiveObjectPermission(ctx, user, "Lead")
	}
	if repo.objPermLoads != 3 {
		t.Fatalf("without cache each load must query; want 3, got %d", repo.objPermLoads)
	}

	repo.objPermLoads = 0
	ctxA := WithAccessInputCache(context.Background())
	_, okA, errA := svc.cachedEffectiveObjectPermission(ctxA, user, "Lead")
	userB := &models.UserSession{ID: "user-1", ProfileID: "other_profile"}
	_, okB, errB := svc.cachedEffectiveObjectPermission(ctxA, userB, "Lead")
	if !okA || errA != nil {
		t.Fatal("first profile load should be served by the memo")
	}
	if !okB || !errors.Is(errB, errAccessInputCachePermissionIdentityMismatch) {
		t.Fatal("profile switch mid-request must fail closed, not serve stale grants")
	}
}

func TestAccessInputCache_ObjectPermissionEmptyProfileDoesNotMatchPopulatedProfile(t *testing.T) {
	grant := &models.ObjectPermissions{SobjectType: "Lead", PermissionsRead: true}
	repo := &countingShareRepo{
		saveTestRepo:     &saveTestRepo{},
		objectPermByName: map[string]*models.ObjectPermissions{"Lead": grant},
	}
	svc := newCacheTestService(repo)
	ctx := WithAccessInputCache(context.Background())

	emptyProfileUser := &models.UserSession{ID: "user-1"}
	if _, ok, err := svc.cachedEffectiveObjectPermission(ctx, emptyProfileUser, "Lead"); !ok || err != nil {
		t.Fatal("empty-profile first load should be served by the memo")
	}

	populatedProfileUser := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	if _, ok, err := svc.cachedEffectiveObjectPermission(ctx, populatedProfileUser, "Lead"); !ok || !errors.Is(err, errAccessInputCachePermissionIdentityMismatch) {
		t.Fatal("empty-profile memo must fail closed for a later populated-profile lookup in the same request")
	}
}

func TestAccessInputCache_ObjectPermissionLoadsDoNotHoldGlobalCacheLock(t *testing.T) {
	repo := &concurrentObjectPermissionRepo{
		saveTestRepo: &saveTestRepo{},
		entered:      make(chan struct{}, 3),
		release:      make(chan struct{}),
	}
	svc := NewPermissionService(repo, &cacheTestMetadata{}, nil)
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	ctx := WithAccessInputCache(context.Background())

	var wg sync.WaitGroup
	errs := make(chan error, 3)
	for _, objectName := range []string{"Account", "Campaign", "User"} {
		objectName := objectName
		wg.Add(1)
		go func() {
			defer wg.Done()
			_, err := svc.loadEffectiveObjectPermission(ctx, user, objectName)
			errs <- err
		}()
	}
	for range 3 {
		select {
		case <-repo.entered:
		case <-time.After(time.Second):
			t.Fatal("object-permission loads serialized behind the request-cache mutex")
		}
	}
	close(repo.release)
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatalf("loadEffectiveObjectPermission: %v", err)
		}
	}
	if got := repo.max.Load(); got != 3 {
		t.Fatalf("max concurrent object-permission loads = %d, want 3", got)
	}

	for _, objectName := range []string{"Account", "Campaign", "User"} {
		if _, err := svc.loadEffectiveObjectPermission(ctx, user, objectName); err != nil {
			t.Fatalf("cached load for %s: %v", objectName, err)
		}
	}
	if got := repo.calls.Load(); got != 3 {
		t.Fatalf("repository calls after cache hits = %d, want 3", got)
	}
}

func TestAccessInputCache_FieldPermissionObjectProjectionCollapsesPointLoads(t *testing.T) {
	explicitDeny := &models.FieldPermissions{
		SobjectType:     "Account",
		Field:           "Account.Secret__c",
		PermissionsRead: true,
		PermissionsEdit: false,
	}
	repo := &countingShareRepo{
		saveTestRepo: &saveTestRepo{},
		objectPermByName: map[string]*models.ObjectPermissions{
			"Account": {SobjectType: "Account", PermissionsRead: true, PermissionsCreate: true, PermissionsEdit: true},
		},
		fieldPermByRef: map[string]*models.FieldPermissions{
			"Account.Secret__c": explicitDeny,
		},
	}
	svc := newCacheTestService(repo)
	user := &models.UserSession{ID: "user-1", ProfileID: "standard_user"}
	ctx := WithAccessInputCache(context.Background())

	if !mustCheckFieldVisibilityWithUser(t, svc, ctx, "Account", "Secret__c", user) {
		t.Fatal("explicit read grant from request field snapshot was not applied")
	}
	if mustCheckFieldEditabilityWithUser(t, svc, ctx, "Account", "Secret__c", user) {
		t.Fatal("explicit edit deny from request field snapshot was not applied")
	}

	for i := 0; i < 3; i++ {
		if !mustCheckFieldVisibilityWithUser(t, svc, ctx, "Account", "Missing__c", user) {
			t.Fatal("missing FLS row should preserve object-read fallback")
		}
		if !mustCheckFieldEditabilityWithUser(t, svc, ctx, "Account", "Missing__c", user) {
			t.Fatal("missing FLS row should preserve object-edit fallback")
		}
	}
	if repo.fieldPermObjectLoads != 1 {
		t.Fatalf("field object projection loads = %d; want 1", repo.fieldPermObjectLoads)
	}
	if repo.fieldPermBulkLoads != 0 {
		t.Fatalf("all-object field bulk loads = %d; want 0", repo.fieldPermBulkLoads)
	}
	if repo.fieldPermLoads != 0 {
		t.Fatalf("point field permission loads = %d; want 0", repo.fieldPermLoads)
	}
}

// boundedFieldProjectionRepo is a request-cache test double with a deliberately
// separate all-object method so tests fail if the cold path regresses to it.
type boundedFieldProjectionRepo struct {
	*saveTestRepo

	mu sync.Mutex

	allObjectCalls int
	objectCalls    int
	pointCalls     int
	objectRows     map[string][]models.FieldPermissions
	allObjectRows  []models.FieldPermissions
	objectErrors   []error
	panicCalls     int
	started        chan struct{}
	release        chan struct{}
	beforeReturn   func()
}

func (r *boundedFieldProjectionRepo) ListEffectiveFieldPermissionsForUser(context.Context, string, string) ([]models.FieldPermissions, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.allObjectCalls++
	return append([]models.FieldPermissions(nil), r.allObjectRows...), nil
}

func (r *boundedFieldProjectionRepo) ListEffectiveFieldPermissionsForUserObject(ctx context.Context, _, _, objectAPIName string) ([]models.FieldPermissions, error) {
	r.mu.Lock()
	call := r.objectCalls
	r.objectCalls++
	rows := append([]models.FieldPermissions(nil), r.objectRows[strings.ToLower(objectAPIName)]...)
	var err error
	if call < len(r.objectErrors) {
		err = r.objectErrors[call]
	}
	started := r.started
	release := r.release
	beforeReturn := r.beforeReturn
	r.mu.Unlock()
	if call < r.panicCalls {
		panic("bounded field projection panic")
	}

	if started != nil {
		select {
		case started <- struct{}{}:
		default:
		}
	}
	if release != nil {
		select {
		case <-release:
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}
	if beforeReturn != nil {
		beforeReturn()
	}
	if err != nil {
		return nil, err
	}
	return rows, nil
}

func (r *boundedFieldProjectionRepo) LoadEffectiveFieldPermission(context.Context, *models.UserSession, string, string) (*models.FieldPermissions, error) {
	r.mu.Lock()
	r.pointCalls++
	r.mu.Unlock()
	return &models.FieldPermissions{PermissionsRead: true, PermissionsEdit: true}, nil
}

func (r *boundedFieldProjectionRepo) calls() (allObject, object int) {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.allObjectCalls, r.objectCalls
}

func (r *boundedFieldProjectionRepo) pointCallCount() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.pointCalls
}

func TestAccessInputCache_BoundedFieldProjectionCoalescesParallelColdLoad(t *testing.T) {
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows: map[string][]models.FieldPermissions{
			strings.ToLower(constants.TableNotification): {
				{SobjectType: constants.TableNotification, Field: "Id", PermissionsRead: true},
				{SobjectType: constants.TableNotification, Field: "Title", PermissionsRead: true},
			},
		},
		started: make(chan struct{}, 1),
		release: make(chan struct{}),
	}
	svc := NewPermissionService(repo, nil, nil, WithFieldProjectionAdmissionLimits(1, 1))
	user := &models.UserSession{ID: "user-1", ProfileID: "profile-1"}
	ctx := WithDiagnostics(WithAccessInputCache(context.Background()), NewDiagnostics())

	const checks = 16
	results := make(chan bool, checks)
	var wg sync.WaitGroup
	for i := 0; i < checks; i++ {
		field := "Id"
		if i%2 == 1 {
			field = "Title"
		}
		wg.Add(1)
		go func(field string) {
			defer wg.Done()
			perm, handled, err := svc.cachedEffectiveFieldPermission(ctx, user, constants.TableNotification, field)
			results <- handled && err == nil && perm != nil && perm.PermissionsRead
		}(field)
	}

	select {
	case <-repo.started:
	case <-time.After(time.Second):
		t.Fatal("bounded field projection did not start")
	}
	close(repo.release)
	wg.Wait()
	close(results)
	for ok := range results {
		if !ok {
			t.Fatal("parallel field check did not receive the bounded projection result")
		}
	}

	allObject, object := repo.calls()
	if allObject != 0 || object != 1 {
		t.Fatalf("projection calls = all-object %d, object %d; want 0, 1", allObject, object)
	}
	diag := SnapshotDiagnostics(ctx)
	if diag.FieldBulkCalls != 1 || diag.FieldBulkRows != 2 {
		t.Fatalf("permission diagnostics = %+v; want one two-row bounded load", diag)
	}
}

func TestAccessInputCache_BoundedFieldProjectionCancellationIsTerminalWithinRequest(t *testing.T) {
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows: map[string][]models.FieldPermissions{
			"account": {{SobjectType: "Account", Field: "Name", PermissionsRead: true}},
		},
		started: make(chan struct{}, 2),
		release: make(chan struct{}),
	}
	svc := NewPermissionService(repo, nil, nil, WithFieldProjectionAdmissionLimits(1, defaultBoundedFieldPermissionProjectionWaiterLimit))
	user := &models.UserSession{ID: "user-1", ProfileID: "profile-1"}
	baseCtx := WithAccessInputCache(context.Background())
	leaderCtx, cancelLeader := context.WithCancel(baseCtx)
	defer cancelLeader()
	leaderResult := make(chan error, 1)
	go func() {
		_, handled, err := svc.cachedEffectiveFieldPermission(leaderCtx, user, "Account", "Name")
		if !handled && err == nil {
			err = errors.New("canceled projection was not handled by request cache")
		}
		leaderResult <- err
	}()
	select {
	case <-repo.started:
	case <-time.After(time.Second):
		t.Fatal("leader projection did not start")
	}

	waiterResult := make(chan error, 1)
	go func() {
		_, handled, err := svc.cachedEffectiveFieldPermission(baseCtx, user, "Account", "Name")
		if !handled && err == nil {
			err = errors.New("canceled projection was not handled by request cache")
		}
		waiterResult <- err
	}()
	cancelLeader()
	if err := <-leaderResult; !errors.Is(err, context.Canceled) {
		t.Fatalf("canceled loader error = %v, want context canceled", err)
	}
	if err := <-waiterResult; !errors.Is(err, context.Canceled) {
		t.Fatalf("waiter error = %v, want cached context canceled", err)
	}
	allObject, object := repo.calls()
	if allObject != 0 || object != 1 {
		t.Fatalf("same-request calls after cancellation = all-object %d, object %d; want 0, 1", allObject, object)
	}

	close(repo.release)
	nextCtx := WithAccessInputCache(context.Background())
	perm, handled, err := svc.cachedEffectiveFieldPermission(nextCtx, user, "Account", "Name")
	if !handled || err != nil || perm == nil || !perm.PermissionsRead {
		t.Fatal("new request did not retry the canceled bounded projection")
	}
	allObject, object = repo.calls()
	if allObject != 0 || object != 2 {
		t.Fatalf("cross-request calls after cancellation = all-object %d, object %d; want 0, 2", allObject, object)
	}
}

func TestAccessInputCache_CancelThenRepositorySuccessDoesNotCacheGrant(t *testing.T) {
	baseCtx := WithAccessInputCache(context.Background())
	loaderCtx, cancel := context.WithCancel(baseCtx)
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows: map[string][]models.FieldPermissions{
			"account": {{SobjectType: "Account", Field: "Name", PermissionsRead: true}},
		},
		beforeReturn: cancel,
	}
	svc := NewPermissionService(repo, nil, nil, WithFieldProjectionAdmissionLimits(1, 0))
	user := &models.UserSession{ID: "user-1", ProfileID: "profile-1"}

	perm, handled, err := svc.cachedEffectiveFieldPermission(loaderCtx, user, "Account", "Name")
	if perm != nil || !handled || !errors.Is(err, context.Canceled) {
		t.Fatalf("canceled loader = perm %#v, handled %v, error %v; want nil,true,context canceled", perm, handled, err)
	}
	perm, handled, err = svc.cachedEffectiveFieldPermission(baseCtx, user, "Account", "Name")
	if perm != nil || !handled || !errors.Is(err, context.Canceled) {
		t.Fatalf("same-request replay = perm %#v, handled %v, error %v; want terminal context canceled", perm, handled, err)
	}
	if active, waiters := svc.fieldProjectionAdmission.snapshot(); active != 0 || waiters != 0 {
		t.Fatalf("canceled-success admission leaked: active=%d waiters=%d", active, waiters)
	}

	repo.mu.Lock()
	repo.beforeReturn = nil
	repo.mu.Unlock()
	perm, handled, err = svc.cachedEffectiveFieldPermission(WithAccessInputCache(context.Background()), user, "Account", "Name")
	if perm == nil || !perm.PermissionsRead || !handled || err != nil {
		t.Fatalf("new-request retry = perm %#v, handled %v, error %v; want fresh grant", perm, handled, err)
	}
	allObject, object := repo.calls()
	if allObject != 0 || object != 2 || repo.pointCallCount() != 0 {
		t.Fatalf("permission calls = all-object %d, object %d, point %d; want 0,2,0", allObject, object, repo.pointCallCount())
	}
}

func TestAccessInputCache_BoundedFieldProjectionFailureIsTerminalWithinRequest(t *testing.T) {
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows: map[string][]models.FieldPermissions{
			"account": {{SobjectType: "Account", Field: "Name", PermissionsRead: true}},
		},
		objectErrors: []error{errors.New("temporary projection failure"), nil},
		started:      make(chan struct{}, 1),
		release:      make(chan struct{}),
	}
	svc := NewPermissionService(repo, nil, nil, WithFieldProjectionAdmissionLimits(1, defaultBoundedFieldPermissionProjectionWaiterLimit))
	user := &models.UserSession{ID: "user-1", ProfileID: "profile-1"}
	ctx := WithAccessInputCache(context.Background())

	const checks = 16
	results := make(chan error, checks)
	var wg sync.WaitGroup
	for i := 0; i < checks; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_, handled, err := svc.cachedEffectiveFieldPermission(ctx, user, "Account", "Name")
			if !handled && err == nil {
				err = errors.New("failed projection was not handled by request cache")
			}
			results <- err
		}()
	}
	select {
	case <-repo.started:
	case <-time.After(time.Second):
		t.Fatal("failed bounded projection did not start")
	}
	close(repo.release)
	wg.Wait()
	close(results)
	for err := range results {
		if err == nil || err.Error() != "temporary projection failure" {
			t.Fatalf("same-request failed projection error = %v", err)
		}
	}
	allObject, object := repo.calls()
	if allObject != 0 || object != 1 {
		t.Fatalf("same-request calls after failure = all-object %d, object %d; want 0, 1", allObject, object)
	}

	nextCtx := WithAccessInputCache(context.Background())
	perm, handled, err := svc.cachedEffectiveFieldPermission(nextCtx, user, "Account", "Name")
	if !handled || err != nil || perm == nil || !perm.PermissionsRead {
		t.Fatal("new request did not retry the failed bounded projection")
	}
	allObject, object = repo.calls()
	if allObject != 0 || object != 2 {
		t.Fatalf("cross-request calls after failure = all-object %d, object %d; want 0, 2", allObject, object)
	}
}

func TestAccessInputCache_BoundedFieldProjectionBindsUserAndProfile(t *testing.T) {
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows: map[string][]models.FieldPermissions{
			"account": {{SobjectType: "Account", Field: "Name", PermissionsRead: true}},
		},
	}
	svc := NewPermissionService(repo, nil, nil)
	ctx := WithAccessInputCache(context.Background())
	owner := &models.UserSession{ID: "user-1", ProfileID: "profile-1"}
	if _, handled, err := svc.cachedEffectiveFieldPermission(ctx, owner, "Account", "Name"); !handled || err != nil {
		t.Fatal("owner did not load bounded projection")
	}
	for _, other := range []*models.UserSession{
		{ID: "user-2", ProfileID: "profile-1"},
		{ID: "user-1", ProfileID: "profile-2"},
	} {
		if _, handled, err := svc.cachedEffectiveFieldPermission(ctx, other, "Account", "Name"); !handled || !errors.Is(err, errAccessInputCachePermissionIdentityMismatch) {
			t.Fatalf("request cache mismatch result for user/profile %+v = handled %v, error %v", other, handled, err)
		}
	}
	allObject, object := repo.calls()
	if allObject != 0 || object != 1 {
		t.Fatalf("projection calls = all-object %d, object %d; want 0, 1", allObject, object)
	}
}

func TestAccessInputCache_BoundedFieldProjectionBindsEmptyProfileExactly(t *testing.T) {
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows: map[string][]models.FieldPermissions{
			"account": {{SobjectType: "Account", Field: "Name", PermissionsRead: true}},
		},
	}
	svc := NewPermissionService(repo, nil, nil)
	ctx := WithAccessInputCache(context.Background())
	if _, handled, err := svc.cachedEffectiveFieldPermission(ctx, &models.UserSession{ID: "user-1"}, "Account", "Name"); !handled || err != nil {
		t.Fatal("empty-profile request did not load its bounded projection")
	}
	if _, handled, err := svc.cachedEffectiveFieldPermission(ctx, &models.UserSession{ID: "user-1", ProfileID: "profile-1"}, "Account", "Name"); !handled || !errors.Is(err, errAccessInputCachePermissionIdentityMismatch) {
		t.Fatalf("empty-to-populated profile mismatch = handled %v, error %v", handled, err)
	}
	allObject, object := repo.calls()
	if allObject != 0 || object != 1 {
		t.Fatalf("projection calls = all-object %d, object %d; want 0, 1", allObject, object)
	}
}

func TestAccessInputCache_BoundedFieldProjectionBindsEmptyUserExactly(t *testing.T) {
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows: map[string][]models.FieldPermissions{
			"account": {{SobjectType: "Account", Field: "Name", PermissionsRead: true}},
		},
	}
	svc := NewPermissionService(repo, nil, nil)
	ctx := WithAccessInputCache(context.Background())
	if _, handled, err := svc.cachedEffectiveFieldPermission(ctx, &models.UserSession{ProfileID: "profile-1"}, "Account", "Name"); !handled || err != nil {
		t.Fatal("empty-user request did not load its bounded projection")
	}
	if _, handled, err := svc.cachedEffectiveFieldPermission(ctx, &models.UserSession{ID: "user-1", ProfileID: "profile-1"}, "Account", "Name"); !handled || !errors.Is(err, errAccessInputCachePermissionIdentityMismatch) {
		t.Fatalf("empty-to-populated user mismatch = handled %v, error %v", handled, err)
	}
	allObject, object := repo.calls()
	if allObject != 0 || object != 1 {
		t.Fatalf("projection calls = all-object %d, object %d; want 0, 1", allObject, object)
	}
}

func TestAccessInputCache_BoundedFieldProjectionPanicIsTerminalWithinRequest(t *testing.T) {
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows: map[string][]models.FieldPermissions{
			"account": {{SobjectType: "Account", Field: "Name", PermissionsRead: true}},
		},
		panicCalls: 1,
	}
	svc := NewPermissionService(repo, nil, nil, WithFieldProjectionAdmissionLimits(1, defaultBoundedFieldPermissionProjectionWaiterLimit))
	user := &models.UserSession{ID: "user-1", ProfileID: "profile-1"}
	ctx := WithDiagnostics(WithAccessInputCache(context.Background()), NewDiagnostics())

	if _, handled, err := svc.cachedEffectiveFieldPermission(ctx, user, "Account", "Name"); !handled || err == nil {
		t.Fatalf("panicked projection handled/error = %v/%v, want true/non-nil", handled, err)
	}
	if _, handled, err := svc.cachedEffectiveFieldPermission(ctx, user, "Account", "Name"); !handled || err == nil {
		t.Fatalf("same-request cached panic handled/error = %v/%v, want true/non-nil", handled, err)
	}
	allObject, object := repo.calls()
	if allObject != 0 || object != 1 {
		t.Fatalf("same-request calls after panic = all-object %d, object %d; want 0, 1", allObject, object)
	}
	if got := SnapshotDiagnostics(ctx).FieldBulkCalls; got != 1 {
		t.Fatalf("same-request diagnostics after panic = %d bulk calls, want 1", got)
	}

	nextCtx := WithAccessInputCache(context.Background())
	perm, handled, err := svc.cachedEffectiveFieldPermission(nextCtx, user, "Account", "Name")
	if !handled || err != nil || perm == nil || !perm.PermissionsRead {
		t.Fatal("new request did not retry after projection panic")
	}
	allObject, object = repo.calls()
	if allObject != 0 || object != 2 {
		t.Fatalf("cross-request calls after panic = all-object %d, object %d; want 0, 2", allObject, object)
	}
}

func TestPermissionService_BoundedFieldProjectionAdmissionWaitsWithinContextAndFailsClosed(t *testing.T) {
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows: map[string][]models.FieldPermissions{
			"account": {{SobjectType: "Account", Field: "Name", PermissionsRead: true}},
		},
		started: make(chan struct{}, 1),
		release: make(chan struct{}),
	}
	svc := NewPermissionService(repo, &pcMetadataStub{schemas: map[string]*models.ObjectMetadata{"Account": {APIName: "Account", Fields: []models.FieldMetadata{{APIName: "Name"}}}}}, nil, WithFieldProjectionAdmissionLimits(1, defaultBoundedFieldPermissionProjectionWaiterLimit))
	firstResult := make(chan error, 1)
	go func() {
		_, err := svc.ListEffectiveFieldPermissionsForUserObject(context.Background(), "user-1", "profile-1", "Account")
		firstResult <- err
	}()
	select {
	case <-repo.started:
	case <-time.After(time.Second):
		t.Fatal("first bounded projection did not occupy admission")
	}

	diagCtx, cancelWait := context.WithCancel(WithAccessInputCache(WithDiagnostics(context.Background(), NewDiagnostics())))
	waitingResult := make(chan error, 1)
	go func() {
		_, err := svc.loadEffectiveFieldPermission(diagCtx, &models.UserSession{ID: "user-2", ProfileID: "profile-2"}, "Account", "Name")
		waitingResult <- err
	}()
	select {
	case err := <-waitingResult:
		t.Fatalf("contended projection returned before capacity or cancellation: %v", err)
	case <-time.After(50 * time.Millisecond):
	}
	if _, got := svc.fieldProjectionAdmission.snapshot(); got != 1 {
		t.Fatalf("registered admission waiters = %d, want 1", got)
	}
	if _, object := repo.calls(); object != 1 {
		t.Fatalf("repository object calls while waiting = %d, want only admitted call", object)
	}
	if calls := repo.pointCallCount(); calls != 0 {
		t.Fatalf("waiting bounded load fell back to %d point queries; want 0", calls)
	}
	cancelWait()
	if err := <-waitingResult; !errors.Is(err, context.Canceled) {
		t.Fatalf("canceled admission wait error = %v, want context canceled", err)
	}
	if got := SnapshotDiagnostics(diagCtx).FieldBulkRejected; got != 1 {
		t.Fatalf("canceled admission diagnostics = %d, want 1", got)
	}
	if _, got := svc.fieldProjectionAdmission.snapshot(); got != 0 {
		t.Fatalf("admission waiters after cancellation = %d, want 0", got)
	}

	canceledCtx, cancel := context.WithCancel(context.Background())
	cancel()
	if _, err := svc.ListEffectiveFieldPermissionsForUserObject(canceledCtx, "user-3", "profile-3", "Account"); !errors.Is(err, context.Canceled) {
		t.Fatalf("canceled projection error = %v, want context canceled", err)
	}

	close(repo.release)
	if err := <-firstResult; err != nil {
		t.Fatalf("admitted projection failed: %v", err)
	}
	_, object := repo.calls()
	if object != 1 {
		t.Fatalf("repository object calls = %d, want only the admitted call", object)
	}
}

func TestPermissionService_BoundedFieldProjectionCancelReleaseRaceDoesNotLeakAdmission(t *testing.T) {
	type acquireResult struct {
		release func()
		err     error
	}

	for i := 0; i < 500; i++ {
		admission := newBoundedFieldProjectionAdmission(1, 1)
		svc := &PermissionService{fieldProjectionAdmission: admission}
		occupiedRelease, err := admission.acquire(context.Background())
		if err != nil {
			t.Fatalf("iteration %d occupy admission: %v", i, err)
		}
		ctx, cancel := context.WithCancel(context.Background())
		result := make(chan acquireResult, 1)
		go func() {
			release, err := svc.acquireBoundedFieldPermissionProjection(ctx)
			result <- acquireResult{release: release, err: err}
		}()

		deadline := time.Now().Add(time.Second)
		for func() bool { _, waiters := admission.snapshot(); return waiters != 1 }() && time.Now().Before(deadline) {
			time.Sleep(time.Microsecond)
		}
		if _, got := admission.snapshot(); got != 1 {
			t.Fatalf("iteration %d registered waiters = %d, want 1", i, got)
		}

		start := make(chan struct{})
		var wg sync.WaitGroup
		wg.Add(2)
		go func() {
			defer wg.Done()
			<-start
			cancel()
		}()
		go func() {
			defer wg.Done()
			<-start
			occupiedRelease()
		}()
		close(start)
		wg.Wait()

		got := <-result
		if got.err != nil && !errors.Is(got.err, context.Canceled) {
			t.Fatalf("iteration %d acquire error = %v, want nil or context canceled", i, got.err)
		}
		if got.release != nil {
			got.release()
		}
		if active, waiters := admission.snapshot(); waiters != 0 || active != 0 {
			t.Fatalf("iteration %d leaked admission: waiters=%d active=%d", i, waiters, active)
		}
	}
}

func TestPermissionService_DefaultBoundedFieldProjectionAdmissionAppliesFiniteBackpressure(t *testing.T) {
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows: map[string][]models.FieldPermissions{
			"account": {{SobjectType: "Account", Field: "Name", PermissionsRead: true}},
		},
		started: make(chan struct{}, defaultBoundedFieldPermissionProjectionLimit+defaultBoundedFieldPermissionProjectionWaiterLimit+1),
		release: make(chan struct{}),
	}
	svc := NewPermissionService(repo, &pcMetadataStub{schemas: map[string]*models.ObjectMetadata{"Account": {APIName: "Account", Fields: []models.FieldMetadata{{APIName: "Name"}}}}}, nil)
	var releaseOnce sync.Once
	releaseAll := func() { releaseOnce.Do(func() { close(repo.release) }) }
	defer releaseAll()

	totalAdmitted := defaultBoundedFieldPermissionProjectionLimit + defaultBoundedFieldPermissionProjectionWaiterLimit
	results := make(chan error, totalAdmitted)
	for i := 0; i < defaultBoundedFieldPermissionProjectionLimit; i++ {
		go func(i int) {
			_, err := svc.ListEffectiveFieldPermissionsForUserObject(context.Background(), fmt.Sprintf("user-%d", i), "profile-1", "Account")
			results <- err
		}(i)
	}
	for i := 0; i < defaultBoundedFieldPermissionProjectionLimit; i++ {
		select {
		case <-repo.started:
		case <-time.After(time.Second):
			t.Fatalf("admitted projection %d did not start", i+1)
		}
	}

	for i := 0; i < defaultBoundedFieldPermissionProjectionWaiterLimit; i++ {
		go func(i int) {
			_, err := svc.ListEffectiveFieldPermissionsForUserObject(context.Background(), fmt.Sprintf("waiting-user-%d", i), "profile-1", "Account")
			results <- err
		}(i)
	}
	waitDeadline := time.Now().Add(time.Second)
	for func() bool {
		_, waiters := svc.fieldProjectionAdmission.snapshot()
		return waiters != defaultBoundedFieldPermissionProjectionWaiterLimit
	}() && time.Now().Before(waitDeadline) {
		time.Sleep(time.Millisecond)
	}
	if _, got := svc.fieldProjectionAdmission.snapshot(); got != defaultBoundedFieldPermissionProjectionWaiterLimit {
		t.Fatalf("registered admission waiters = %d, want finite limit %d", got, defaultBoundedFieldPermissionProjectionWaiterLimit)
	}
	if allObject, object := repo.calls(); allObject != 0 || object != defaultBoundedFieldPermissionProjectionLimit {
		t.Fatalf("repository calls with full wait queue = all-object %d, object %d; want 0, %d", allObject, object, defaultBoundedFieldPermissionProjectionLimit)
	}

	diagCtx := WithAccessInputCache(WithDiagnostics(context.Background(), NewDiagnostics()))
	overflowUser := &models.UserSession{ID: "overflow-user", ProfileID: "profile-1"}
	if _, err := svc.loadEffectiveFieldPermission(diagCtx, overflowUser, "Account", "Name"); !errors.Is(err, ErrFieldPermissionProjectionOverloaded) {
		t.Fatalf("overflow projection error = %v, want finite-admission rejection", err)
	}
	if allowed, err := svc.CheckFieldVisibilityWithUser(diagCtx, "Account", "Name", overflowUser); allowed || !errors.Is(err, ErrFieldPermissionProjectionOverloaded) {
		t.Fatalf("same-request overflow decision = allowed %v, error %v; want false, typed retriable error", allowed, err)
	}
	overflowDiag := SnapshotDiagnostics(diagCtx)
	if overflowDiag.FieldBulkRejected != 1 || overflowDiag.FieldBulkCalls != 0 {
		t.Fatalf("same-request overflow diagnostics = %+v; want one rejection and zero repository bulk calls", overflowDiag)
	}
	if calls := repo.pointCallCount(); calls != 0 {
		t.Fatalf("overflow bounded load fell back to %d point queries; want 0", calls)
	}
	if allObject, object := repo.calls(); allObject != 0 || object != defaultBoundedFieldPermissionProjectionLimit {
		t.Fatalf("repository calls after overflow = all-object %d, object %d; want unchanged 0, %d", allObject, object, defaultBoundedFieldPermissionProjectionLimit)
	}

	releaseAll()
	for i := 0; i < totalAdmitted; i++ {
		select {
		case err := <-results:
			if err != nil {
				t.Fatalf("admitted projection failed: %v", err)
			}
		case <-time.After(time.Second):
			t.Fatalf("admitted projection %d/%d did not finish", i+1, totalAdmitted)
		}
	}
	if _, got := svc.fieldProjectionAdmission.snapshot(); got != 0 {
		t.Fatalf("admission waiters after release = %d, want 0", got)
	}
	if got, _ := svc.fieldProjectionAdmission.snapshot(); got != 0 {
		t.Fatalf("admission active loads after release = %d, want 0", got)
	}
	nextCtx := WithAccessInputCache(context.Background())
	if !mustCheckFieldVisibilityWithUser(t, svc, nextCtx, "Account", "Name", overflowUser) {
		t.Fatal("new request did not retry authority after admission rejection")
	}
	_, object := repo.calls()
	if object != totalAdmitted+1 {
		t.Fatalf("repository object calls = %d, want %d admitted plus one new-request retry", object, totalAdmitted+1)
	}
}

func TestAccessInputCache_BoundedFieldProjectionHasNoCrossRequestRevocationWindow(t *testing.T) {
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows: map[string][]models.FieldPermissions{
			"account": {{SobjectType: "Account", Field: "Name", PermissionsRead: true}},
		},
	}
	svc := NewPermissionService(repo, nil, nil)
	user := &models.UserSession{ID: "user-1", ProfileID: "profile-1"}
	firstCtx := WithAccessInputCache(context.Background())
	if perm, handled, err := svc.cachedEffectiveFieldPermission(firstCtx, user, "Account", "Name"); !handled || err != nil || perm == nil || !perm.PermissionsRead {
		t.Fatal("first request did not receive the initial grant")
	}

	repo.mu.Lock()
	repo.objectRows["account"] = []models.FieldPermissions{{SobjectType: "Account", Field: "Name", PermissionsRead: false}}
	repo.mu.Unlock()
	secondCtx := WithAccessInputCache(context.Background())
	if perm, handled, err := svc.cachedEffectiveFieldPermission(secondCtx, user, "Account", "Name"); !handled || err != nil || perm == nil || perm.PermissionsRead {
		t.Fatal("new request reused a stale field grant after revocation")
	}
	allObject, object := repo.calls()
	if allObject != 0 || object != 2 {
		t.Fatalf("projection calls across requests = all-object %d, object %d; want 0, 2", allObject, object)
	}
}

func TestAccessInputCache_NotificationRouteShapeUsesBoundedProjection(t *testing.T) {
	notificationFields := []models.FieldPermissions{
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_ID, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_RecipientID, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_Title, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_Body, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_Link, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_IsRead, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_NotificationType, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_CreatedDate, PermissionsRead: true},
	}
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows:   map[string][]models.FieldPermissions{strings.ToLower(constants.TableNotification): notificationFields},
	}
	svc := NewPermissionService(repo, nil, nil)
	user := &models.UserSession{ID: "user-1", ProfileID: "profile-1"}
	ctx := WithDiagnostics(WithAccessInputCache(context.Background()), NewDiagnostics())
	for _, field := range notificationFields {
		perm, handled, err := svc.cachedEffectiveFieldPermission(ctx, user, constants.TableNotification, field.Field)
		if !handled || err != nil || perm == nil || !perm.PermissionsRead {
			t.Fatalf("notification field %s was not granted by the bounded projection", field.Field)
		}
	}
	allObject, object := repo.calls()
	if allObject != 0 || object != 1 {
		t.Fatalf("notification route calls = all-object %d, object %d; want 0, 1", allObject, object)
	}
	diag := SnapshotDiagnostics(ctx)
	if diag.FieldBulkCalls != 1 || diag.FieldBulkRows != len(notificationFields) {
		t.Fatalf("notification route diagnostics = %+v; want one %d-row projection", diag, len(notificationFields))
	}
}

func BenchmarkAccessInputCache_ColdNotificationBoundedProjection(b *testing.B) {
	notificationFields := []models.FieldPermissions{
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_ID, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_RecipientID, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_Title, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_Body, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_Link, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_IsRead, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_NotificationType, PermissionsRead: true},
		{SobjectType: constants.TableNotification, Field: constants.FieldNotification_CreatedDate, PermissionsRead: true},
	}
	repo := &boundedFieldProjectionRepo{
		saveTestRepo: &saveTestRepo{},
		objectRows:   map[string][]models.FieldPermissions{strings.ToLower(constants.TableNotification): notificationFields},
	}
	svc := NewPermissionService(repo, nil, nil)
	user := &models.UserSession{ID: "user-1", ProfileID: "profile-1"}

	b.ReportAllocs()
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		ctx := WithAccessInputCache(context.Background())
		for _, field := range notificationFields {
			perm, handled, err := svc.cachedEffectiveFieldPermission(ctx, user, constants.TableNotification, field.Field)
			if !handled || err != nil || perm == nil || !perm.PermissionsRead {
				b.Fatalf("notification field %s was not granted by bounded projection", field.Field)
			}
		}
	}
	b.StopTimer()
	allObject, object := repo.calls()
	if allObject != 0 || object != b.N {
		b.Fatalf("benchmark calls = all-object %d, object %d; want 0, %d", allObject, object, b.N)
	}
}