package permissions

import (
	"context"
	"encoding/json"
	"fmt"
	"slices"
	"sort"
	"strings"
	"unicode/utf8"

	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	cp "github.com/pingcap-inc/nexuscrm-contract/pkg/controlplane"
)

// AssignmentPolicyEligibilityState is a product-neutral eligibility image
// supplied by an adapter after the adapter has interpreted its record model.
// Selectors are declaration keys; an empty set means the source is not
// currently eligible for any selector-backed policy.
type AssignmentPolicyEligibilityState struct {
	Selectors []string `json:"selectors,omitempty"`
}

// AssignmentPolicyEligibilityMutation carries one adapter-owned transition.
// SourceIdentity is an opaque, byte-exact valid UTF-8 identity; the permissions
// engine never trims, folds, or interprets it. ReconcileAfterSelectors tells
// the engine that the population represented by the after selectors may have
// changed even when the selector keys themselves did not; every affected target
// is therefore reconciled to its exact eligible population in the caller-owned
// transaction.
type AssignmentPolicyEligibilityMutation struct {
	SourceIdentity          string                           `json:"sourceIdentity"`
	Before                  AssignmentPolicyEligibilityState `json:"before"`
	After                   AssignmentPolicyEligibilityState `json:"after"`
	ReconcileAfterSelectors bool                             `json:"reconcileAfterSelectors,omitempty"`
}

// AssignmentPolicyEligibilityMutationHandler stages one product-neutral
// eligibility transition in the caller-owned transaction represented by ctx.
type AssignmentPolicyEligibilityMutationHandler interface {
	ApplyAssignmentPolicyEligibilityMutation(context.Context, AssignmentPolicyEligibilityMutation) error
}

// AssignmentPolicyEligibilityMutationBatchHandler stages every eligibility
// transition accumulated by one caller-owned transaction as a single lock
// union.
type AssignmentPolicyEligibilityMutationBatchHandler interface {
	AssignmentPolicyEligibilityMutationHandler
	ApplyAssignmentPolicyEligibilityMutations(context.Context, []AssignmentPolicyEligibilityMutation) error
}

type assignmentPolicySourcePlan struct {
	Mutations       []AssignmentPolicyEligibilityMutation `json:"mutations"`
	SelectorMapping []assignmentPolicySourceSelector      `json:"selectorMapping"`
	LossTargets     []assignmentPolicySourceTarget        `json:"lossTargets"`
	GainTargets     []assignmentPolicySourceTarget        `json:"gainTargets"`
}

type assignmentPolicySourceSelector struct {
	Selector         string   `json:"selector"`
	PermissionSetIDs []string `json:"permissionSetIds"`
}

type assignmentPolicySourceTarget struct {
	PermissionSetID      string `json:"permissionSetId"`
	PermissionSet        string `json:"permissionSet"`
	DefinitionOwnerPack  string `json:"definitionOwnerPack"`
	OwnerPack            string `json:"ownerPack"`
	PolicyKey            string `json:"policyKey"`
	AssignmentRows       int    `json:"assignmentRows,omitempty"`
	SelectorRows         int    `json:"selectorRows,omitempty"`
	EligibilityReadItems int    `json:"eligibilityReadItems,omitempty"`
	EligibilitySubjects  int    `json:"eligibilitySubjects,omitempty"`
}

// NewPermissionSetAssignmentPolicyEligibilityTransitionProvider stages one
// product-neutral eligibility transition. Loss and gain targets are discovered
// and locked as one union, then each target is reconciled to its exact eligible
// population before the caller-owned transaction can commit.
func NewPermissionSetAssignmentPolicyEligibilityTransitionProvider(
	mutation AssignmentPolicyEligibilityMutation,
	eligibility AssignmentPolicyEligibilityProjectionProvider,
) cp.ComponentManagedPlannedTransactionOperationProvider {
	return NewPermissionSetAssignmentPolicyEligibilityTransitionsProvider(
		[]AssignmentPolicyEligibilityMutation{mutation},
		eligibility,
	)
}

// NewPermissionSetAssignmentPolicyEligibilityTransitionsProvider stages a
// complete adapter batch through the same provider engine as a single
// transition. The canonical source list is planned after every adapter-owned
// row in the outer transaction is locked/staged, and all affected physical
// targets are acquired as one union before any policy mutation.
func NewPermissionSetAssignmentPolicyEligibilityTransitionsProvider(
	mutations []AssignmentPolicyEligibilityMutation,
	eligibility AssignmentPolicyEligibilityProjectionProvider,
) cp.ComponentManagedPlannedTransactionOperationProvider {
	frozen := make([]AssignmentPolicyEligibilityMutation, len(mutations))
	for index, mutation := range mutations {
		mutation.Before.Selectors = append([]string(nil), mutation.Before.Selectors...)
		mutation.After.Selectors = append([]string(nil), mutation.After.Selectors...)
		frozen[index] = mutation
	}
	return cp.ComponentManagedPlannedTransactionOperationProvider{
		Plan: func(ctx context.Context, store cp.ComponentManagedTransactionReadStore) (cp.ComponentManagedPlan, error) {
			return planAssignmentPolicySourceTransitions(ctx, store, frozen, eligibility)
		},
		Mutate: func(ctx context.Context, plan cp.ComponentManagedPlan, store cp.ComponentManagedTransactionMutationStore) (cp.ComponentManagedMutationResult, error) {
			return mutateAssignmentPolicySourceTransition(ctx, plan, store, eligibility)
		},
		Project: func(ctx context.Context, plan cp.ComponentManagedPlan, store cp.ComponentManagedTransactionReadStore) (cp.ComponentManagedProjection, error) {
			return projectAssignmentPolicySourceTransition(ctx, plan, store, eligibility)
		},
	}
}

func planAssignmentPolicySourceTransitions(
	ctx context.Context,
	store cp.ComponentManagedReadStore,
	mutations []AssignmentPolicyEligibilityMutation,
	eligibility AssignmentPolicyEligibilityProjectionProvider,
) (cp.ComponentManagedPlan, error) {
	canonical, err := canonicalAssignmentPolicySourceMutations(mutations)
	if err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	planner := newAssignmentPolicySourcePlanner(store)
	lossSet := make(map[string]assignmentPolicySourceTarget)
	gainSet := make(map[string]assignmentPolicySourceTarget)
	lockSet := make(map[string]assignmentPolicySourceTarget)
	activeMutations := make([]AssignmentPolicyEligibilityMutation, 0, len(canonical))
	for _, mutation := range canonical {
		if assignmentPolicyEligibilityStatesEqual(mutation.Before, mutation.After) &&
			!mutation.ReconcileAfterSelectors {
			continue
		}
		beforeTargets, resolveErr := planner.targetsForSelectors(ctx, mutation.Before.Selectors)
		if resolveErr != nil {
			return cp.ComponentManagedPlan{}, resolveErr
		}
		afterTargets, resolveErr := planner.targetsForSelectors(ctx, mutation.After.Selectors)
		if resolveErr != nil {
			return cp.ComponentManagedPlan{}, resolveErr
		}
		for id, target := range beforeTargets {
			lockSet[id] = target
			if _, remains := afterTargets[id]; !remains || mutation.ReconcileAfterSelectors {
				lossSet[id] = target
			}
		}
		for id, target := range afterTargets {
			lockSet[id] = target
			if _, already := beforeTargets[id]; !already {
				gainSet[id] = target
			}
			if mutation.ReconcileAfterSelectors {
				lossSet[id] = target
				gainSet[id] = target
			}
		}
		activeMutations = append(activeMutations, mutation)
	}
	lossTargets := canonicalAssignmentPolicySourceTargets(lossSet)
	gainTargets := canonicalAssignmentPolicySourceTargets(gainSet)
	lockTargets := canonicalAssignmentPolicySourceTargets(lockSet)
	selectorMapping := planner.canonicalSelectorMapping()
	if len(activeMutations) == 0 || (len(lockTargets) == 0 && len(selectorMapping) == 0) {
		return cp.ComponentManagedPlan{}, cp.ErrComponentManagedNoTargets
	}
	maximumCurrentItems := assignmentPolicyManagedBudget().MaxCurrentItems
	selectorMappingRows := 0
	for _, binding := range selectorMapping {
		selectorMappingRows += len(binding.PermissionSetIDs)
	}
	// Planning has already read every selector mapping row and one Definition
	// root plus one authority claim for each physical target. Track those rows
	// before inspecting target populations so Plan itself also fails closed at
	// the registration's current-item budget.
	currentItems := selectorMappingRows + 2*len(lockTargets)
	if currentItems > maximumCurrentItems {
		return cp.ComponentManagedPlan{}, fmt.Errorf(
			"assignment policy eligibility transition current items %d exceed registration budget %d",
			currentItems, maximumCurrentItems,
		)
	}
	observedTargets := make(map[string]assignmentPolicySourceTarget, len(lockTargets))
	for index := range lockTargets {
		assignmentRows, exceeded, countErr := countAssignmentPolicySourceTargetAssignments(
			ctx,
			store,
			lockTargets[index].PermissionSetID,
			maximumCurrentItems-currentItems,
		)
		if countErr != nil {
			return cp.ComponentManagedPlan{}, countErr
		}
		currentItems += assignmentRows
		if exceeded {
			return cp.ComponentManagedPlan{}, fmt.Errorf(
				"assignment policy eligibility transition current items %d exceed registration budget %d",
				currentItems, maximumCurrentItems,
			)
		}
		lockTargets[index].AssignmentRows = assignmentRows
		selectorLimit := min(
			maximumCurrentItems-currentItems,
			assignmentPolicyMaximumRows,
		)
		selectors, selectorPopulationExceeded, selectorErr :=
			readAssignmentPolicySourceTargetSelectorsUpTo(
				ctx,
				store,
				lockTargets[index].PermissionSetID,
				selectorLimit,
			)
		if selectorErr != nil {
			return cp.ComponentManagedPlan{}, selectorErr
		}
		currentItems += len(selectors)
		if selectorPopulationExceeded {
			if selectorLimit < assignmentPolicyMaximumRows {
				return cp.ComponentManagedPlan{}, fmt.Errorf(
					"assignment policy eligibility transition current items %d exceed registration budget %d",
					currentItems, maximumCurrentItems,
				)
			}
			return cp.ComponentManagedPlan{}, fmt.Errorf(
				"assignment policy source target %s has more than %d eligibility selectors",
				lockTargets[index].PermissionSetID, assignmentPolicyMaximumRows,
			)
		}
		lockTargets[index].SelectorRows = len(selectors)
		eligibilityBudget := newAssignmentPolicyEligibilityReadBudget(maximumCurrentItems - currentItems)
		_, _, eligibleSubjects, observedEligibilityRows, populationErr := projectAssignmentPolicyEligibility(
			ctx, store, lockTargets[index].PermissionSetID, selectors, eligibility, eligibilityBudget,
		)
		if populationErr != nil {
			return cp.ComponentManagedPlan{}, populationErr
		}
		// Ambient planning can run before the caller stages the source rows. At
		// most one row per canonical source transition can enter a target's User
		// population before the locked replan/mutation, so reserve that bounded
		// growth in both the host estimate and the locked population check.
		eligibilityReadItems := observedEligibilityRows - len(eligibleSubjects)
		eligibilityLimit := len(eligibleSubjects) + len(activeMutations)
		currentItems += eligibilityReadItems + eligibilityLimit
		if currentItems > maximumCurrentItems {
			return cp.ComponentManagedPlan{}, fmt.Errorf(
				"assignment policy eligibility transition current items %d exceed registration budget %d",
				currentItems, maximumCurrentItems,
			)
		}
		lockTargets[index].EligibilityReadItems = eligibilityReadItems
		lockTargets[index].EligibilitySubjects = eligibilityLimit
		observedTargets[lockTargets[index].PermissionSetID] = lockTargets[index]
	}
	for index := range lossTargets {
		lossTargets[index] = observedTargets[lossTargets[index].PermissionSetID]
	}
	for index := range gainTargets {
		gainTargets[index] = observedTargets[gainTargets[index].PermissionSetID]
	}
	// The ambient caller can prepare this marker before opening its source
	// transaction. Selector and policy targets are deliberately absent: they are
	// discovered from a view that may precede the global lock, so including them
	// would either require writes after source locks or make the locked target
	// envelope drift when the first policy for a selector commits concurrently.
	managedTargets := []cp.ComponentManagedTarget{
		PermissionSetAssignmentPolicySerializationTarget(),
	}
	opaque, err := json.Marshal(assignmentPolicySourcePlan{
		Mutations: activeMutations, SelectorMapping: selectorMapping,
		LossTargets: lossTargets, GainTargets: gainTargets,
	})
	if err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	// Keep the effect envelope stable across the locked replan. A selector may
	// have no policy in the pre-lock view and acquire one before the global target
	// is locked; the disposition still prevents an effect call for a true no-op.
	effects := assignmentPolicyEffects()
	estimate := estimateAssignmentPolicySourceTransition(
		lossTargets, gainTargets, selectorMappingRows,
	)
	if err := validateAssignmentPolicySourceTransitionEstimate(estimate); err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	if len(managedTargets) > assignmentPolicyManagedBudget().MaxTargets {
		return cp.ComponentManagedPlan{}, fmt.Errorf(
			"assignment policy eligibility transition targets %d exceed registration budget %d",
			len(managedTargets), assignmentPolicyManagedBudget().MaxTargets,
		)
	}
	return cp.NewComponentManagedPlan(cp.ComponentManagedPlanSpec{
		Targets:              managedTargets,
		RequiredCapabilities: assignmentPolicyCapabilities(),
		Estimate:             estimate,
		Effects:              effects,
		ProjectionIdentity: []byte(
			string(PermissionSetAssignmentPolicyComponentKind) + "\x00source\x00" +
				"transition-batch\x00" + assignmentPolicySourceMutationIdentity(activeMutations),
		),
		OpaquePlan: opaque,
	})
}

func estimateAssignmentPolicySourceTransition(
	lossTargets []assignmentPolicySourceTarget,
	gainTargets []assignmentPolicySourceTarget,
	selectorMappingRows int,
) cp.ComponentManagedUsage {
	unionTargets := assignmentPolicySourceTargetUnion(lossTargets, gainTargets)
	// Every existing assignment can require both an assignment mutation and a
	// tombstone-receipt mutation. Every eligible source can require a new
	// assignment, and every affected target requires one settled state update.
	// These are upper bounds, not optimistic expected counts, so the host rejects
	// an oversized union before mutation begins.
	unionAssignmentRows := 0
	unionSelectorRows := 0
	unionEligibilityItems := 0
	unionEligibleAssignments := 0
	for _, target := range unionTargets {
		unionAssignmentRows += target.AssignmentRows
		unionSelectorRows += target.SelectorRows
		unionEligibilityItems += target.EligibilityReadItems + target.EligibilitySubjects
		unionEligibleAssignments += target.EligibilitySubjects
	}
	// The host meters the ambient Plan, the locked replan, Mutate, and Project
	// through one receipt. The exact safe upper bound is:
	//
	//   selector mapping: ambient Plan + locked Plan + Mutate
	//   root/claim/state: 2*(root+claim) in Plan + all three in Mutate/Project
	//   every assignment: two Plan counts + one Project snapshot
	//   assignments: PSA lock + tombstone-event/receipt locks in the worst case
	//   every selector row: two Plan counts + one Project snapshot
	//   selectors: one additional Mutate read
	//   eligibility: two Plans + Mutate locks + Project reads
	//   negative edges: at most one returned row in Mutate and Project per target
	currentRows := 3*selectorMappingRows +
		10*len(unionTargets) +
		6*unionAssignmentRows +
		4*unionSelectorRows +
		4*unionEligibilityItems + 2*len(unionTargets)
	mutationRows := 2*unionAssignmentRows + unionEligibleAssignments + len(unionTargets)
	return cp.ComponentManagedUsage{
		CurrentItems:       currentRows,
		DesiredItems:       unionEligibleAssignments,
		DeltaRows:          mutationRows,
		MutationStatements: mutationRows,
		MutationBytes:      mutationRows * assignmentPolicyMutationBytesPerRow,
		BatchRows:          min(mutationRows, assignmentPolicyMaximumRows),
	}
}

func validateAssignmentPolicySourceTransitionEstimate(usage cp.ComponentManagedUsage) error {
	budget := assignmentPolicyManagedBudget()
	limits := []struct {
		name          string
		actual, limit int
	}{
		{name: "current items", actual: usage.CurrentItems, limit: budget.MaxCurrentItems},
		{name: "desired items", actual: usage.DesiredItems, limit: budget.MaxDesiredItems},
		{name: "delta rows", actual: usage.DeltaRows, limit: budget.MaxDeltaRows},
		{name: "mutation bytes", actual: usage.MutationBytes, limit: budget.MaxMutationBytes},
		{name: "mutation statements", actual: usage.MutationStatements, limit: budget.MaxMutationStatements},
		{name: "batch rows", actual: usage.BatchRows, limit: budget.MaxBatchRows},
	}
	for _, entry := range limits {
		if entry.actual > entry.limit {
			return fmt.Errorf(
				"assignment policy eligibility transition %s %d exceeds registration budget %d",
				entry.name, entry.actual, entry.limit,
			)
		}
	}
	return nil
}

func mutateAssignmentPolicySourceTransition(
	ctx context.Context,
	plan cp.ComponentManagedPlan,
	store cp.ComponentManagedTransactionMutationStore,
	eligibilityProvider AssignmentPolicyEligibilityProjectionProvider,
) (cp.ComponentManagedMutationResult, error) {
	decoded, err := decodeAssignmentPolicySourcePlan(plan)
	if err != nil {
		return cp.ComponentManagedMutationResult{}, err
	}
	if err := lockAndValidateAssignmentPolicySourceSelectorMapping(
		ctx, store, decoded.SelectorMapping,
	); err != nil {
		return cp.ComponentManagedMutationResult{}, err
	}
	union := assignmentPolicySourceTargetUnion(decoded.LossTargets, decoded.GainTargets)
	roots := make([]assignmentPolicyRootLock, 0, len(union))
	for _, target := range union {
		roots = append(roots, assignmentPolicyRootLock{
			PermissionSetID: target.PermissionSetID,
			PermissionSet:   target.PermissionSet,
			OwnerPack:       target.DefinitionOwnerPack,
		})
	}
	if err := lockAssignmentPolicyRoots(ctx, store, roots); err != nil {
		return cp.ComponentManagedMutationResult{}, err
	}
	assignments := assignmentPolicySourceTargetAssignmentRows(union)
	eligibility := assignmentPolicySourceTargetEligibilityBudgetItems(union)
	changed := 0
	for _, target := range union {
		claimRows, lockErr := store.LockRows(ctx, assignmentPolicyQuery(
			schemacatalog.AssignmentPolicyClaimTable,
			[]string{"Id", "PermissionSetId", "OwnerPack", "PolicyKey"}, cp.ComponentRow{"PermissionSetId": target.PermissionSetID},
			1, []string{"PermissionSetId"},
		))
		if lockErr != nil {
			return cp.ComponentManagedMutationResult{}, lockErr
		}
		if len(claimRows) != 1 {
			return cp.ComponentManagedMutationResult{}, fmt.Errorf(
				"assignment policy source target %s has no unique authority claim",
				target.PermissionSetID,
			)
		}
		owner, rowErr := assignmentPolicyRequiredString(claimRows[0], "OwnerPack")
		if rowErr != nil {
			return cp.ComponentManagedMutationResult{}, rowErr
		}
		key, rowErr := assignmentPolicyRequiredString(claimRows[0], "PolicyKey")
		if rowErr != nil {
			return cp.ComponentManagedMutationResult{}, rowErr
		}
		if owner != target.OwnerPack || key != target.PolicyKey {
			return cp.ComponentManagedMutationResult{}, fmt.Errorf("assignment policy source authority changed during transition")
		}
		stateRows, lockErr := store.LockRows(ctx, assignmentPolicyQuery(
			schemacatalog.AssignmentPolicyTargetStateTable,
			[]string{
				"Id", "TargetDomain", "PhysicalKey", "PermissionSetId", "Suspended",
				"WorkGeneration", "AppliedGeneration", "ResolutionState",
			},
			cp.ComponentRow{"PermissionSetId": target.PermissionSetID}, 1, []string{"PermissionSetId"},
		))
		if lockErr != nil {
			return cp.ComponentManagedMutationResult{}, lockErr
		}
		if len(stateRows) != 1 {
			return cp.ComponentManagedMutationResult{}, fmt.Errorf("assignment policy source target state is missing")
		}
		if rowErr := assignmentPolicyValidateTargetStateIdentity(
			stateRows[0], target.PermissionSetID, target.PermissionSet,
		); rowErr != nil {
			return cp.ComponentManagedMutationResult{}, rowErr
		}
		state, rowErr := assignmentPolicyStateFromRow(stateRows[0])
		if rowErr != nil {
			return cp.ComponentManagedMutationResult{}, rowErr
		}
		plannedEligibilityItems := eligibility[target.PermissionSetID]
		profiles, readErr := readAssignmentPolicySourceTargetSelectors(
			ctx, store, target.PermissionSetID,
		)
		if readErr != nil {
			return cp.ComponentManagedMutationResult{}, readErr
		}
		eligibilityBudget := newAssignmentPolicyEligibilityReadBudget(plannedEligibilityItems)
		resolution, _, eligible, _, resolveErr := projectAssignmentPolicyEligibility(
			ctx, store, target.PermissionSetID, profiles, eligibilityProvider, eligibilityBudget,
		)
		if observed, budgetErr := eligibilityBudget.observation(); budgetErr != nil {
			return cp.ComponentManagedMutationResult{}, fmt.Errorf(
				"assignment policy source target %s eligibility population grew from %d to at least %d after planning: %w",
				target.PermissionSetID, plannedEligibilityItems, observed, budgetErr,
			)
		}
		if resolveErr != nil {
			return cp.ComponentManagedMutationResult{}, resolveErr
		}
		if state.Suspended || resolution != assignmentPolicyResolutionResolved {
			eligible = nil
		}
		generation := state.WorkGeneration
		if state.WorkGeneration == state.AppliedGeneration {
			generation, resolveErr = nextAssignmentPolicyGeneration(state.WorkGeneration)
			if resolveErr != nil {
				return cp.ComponentManagedMutationResult{}, resolveErr
			}
		}
		targetChanged, reconcileErr := reconcileAssignmentPolicySourceTargetAssignments(
			ctx,
			store,
			target.OwnerPack,
			target.PolicyKey,
			target.PermissionSetID,
			generation,
			eligible,
			assignments[target.PermissionSetID],
		)
		if reconcileErr != nil {
			return cp.ComponentManagedMutationResult{}, reconcileErr
		}
		values := cp.ComponentRow{
			"WorkGeneration": generation, "AppliedGeneration": generation,
			"ResolutionState": resolution, "ResolutionDigest": assignmentPolicyDigest(resolution),
			"NextAttemptAt": nil, "RetryCount": uint64(0),
			"LastFailureCode": nil, "LastFailureDigest": nil,
		}
		updated, updateErr := store.UpdateRows(ctx, cp.ComponentManagedUpdate{
			Table:                   schemacatalog.AssignmentPolicyTargetStateTable,
			Match:                   cp.ComponentRow{"Id": state.ID},
			Values:                  values,
			CurrentTimestampColumns: []string{"LastModifiedDate"},
			MaximumAffectedRows:     1,
		})
		if err := assignmentPolicyRequireExactWrite(
			"update", schemacatalog.AssignmentPolicyTargetStateTable, updated, updateErr,
		); err != nil {
			return cp.ComponentManagedMutationResult{}, err
		}
		changed += targetChanged + 1
	}
	if changed == 0 {
		return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
	}
	return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationSafeProjectionStaged}, nil
}

func projectAssignmentPolicySourceTransition(
	ctx context.Context,
	plan cp.ComponentManagedPlan,
	store cp.ComponentManagedTransactionReadStore,
	eligibilityProvider AssignmentPolicyEligibilityProjectionProvider,
) (cp.ComponentManagedProjection, error) {
	decoded, err := decodeAssignmentPolicySourcePlan(plan)
	if err != nil {
		return cp.ComponentManagedProjection{}, err
	}
	type targetProjection struct {
		PermissionSet     string `json:"permissionSet"`
		WorkGeneration    uint64 `json:"workGeneration"`
		AppliedGeneration uint64 `json:"appliedGeneration"`
		AssignmentsExact  bool   `json:"assignmentsExact"`
		GenerationSettled bool   `json:"generationSettled"`
	}
	union := assignmentPolicySourceTargetUnion(decoded.LossTargets, decoded.GainTargets)
	projected := make([]targetProjection, 0, len(union))
	for _, target := range union {
		snapshot, readErr := readAssignmentPolicyTarget(ctx, store, target.PermissionSet)
		if readErr != nil {
			return cp.ComponentManagedProjection{}, readErr
		}
		if snapshot.State == nil {
			return cp.ComponentManagedProjection{}, fmt.Errorf(
				"assignment policy source target %s has no durable state",
				target.PermissionSetID,
			)
		}
		resolution, _, eligible, _, resolveErr := projectAssignmentPolicyEligibility(
			ctx,
			store,
			target.PermissionSetID,
			snapshot.Profiles,
			eligibilityProvider,
			newAssignmentPolicyEligibilityReadBudget(assignmentPolicyManagedBudget().MaxCurrentItems),
		)
		if resolveErr != nil {
			return cp.ComponentManagedProjection{}, resolveErr
		}
		if snapshot.State.Suspended || resolution != assignmentPolicyResolutionResolved {
			eligible = nil
		}
		want := make(map[string]struct{}, len(eligible))
		for _, id := range eligible {
			want[assignmentPolicyDatabaseIdentity(id)] = struct{}{}
		}
		exact := true
		for id, row := range snapshot.Assignments {
			if row.IsDeleted {
				continue
			}
			if _, desired := want[id]; !desired {
				exact = false
				continue
			}
			delete(want, id)
		}
		if len(want) != 0 {
			exact = false
		}
		projected = append(projected, targetProjection{
			PermissionSet:     target.PermissionSet,
			WorkGeneration:    snapshot.State.WorkGeneration,
			AppliedGeneration: snapshot.State.AppliedGeneration,
			AssignmentsExact:  exact,
			GenerationSettled: snapshot.State.WorkGeneration == snapshot.State.AppliedGeneration,
		})
	}
	live, _ := json.Marshal(projected)
	expectedProjection := append(make([]targetProjection, 0, len(projected)), projected...)
	for index := range expectedProjection {
		expectedProjection[index].AssignmentsExact = true
		expectedProjection[index].GenerationSettled = true
	}
	expected, _ := json.Marshal(expectedProjection)
	return cp.ComponentManagedProjection{
		Identity: plan.ProjectionIdentity(), Expected: expected, Live: live,
	}, nil
}

func validateAssignmentPolicySourceMutation(mutation AssignmentPolicyEligibilityMutation) error {
	if strings.TrimSpace(mutation.SourceIdentity) == "" {
		return fmt.Errorf("assignment policy eligibility mutation requires a source identity")
	}
	if len(mutation.SourceIdentity) > cp.ComponentManagedMaximumPhysicalKey {
		return fmt.Errorf(
			"assignment policy eligibility source identity exceeds %d bytes",
			cp.ComponentManagedMaximumPhysicalKey,
		)
	}
	if !utf8.ValidString(mutation.SourceIdentity) {
		return fmt.Errorf("assignment policy eligibility source identity must be valid UTF-8")
	}
	if len(mutation.Before.Selectors) > assignmentPolicyMaximumRows ||
		len(mutation.After.Selectors) > assignmentPolicyMaximumRows {
		return fmt.Errorf(
			"assignment policy eligibility mutation exceeds the maximum of %d selectors per state",
			assignmentPolicyMaximumRows,
		)
	}
	for _, state := range []AssignmentPolicyEligibilityState{mutation.Before, mutation.After} {
		for _, selector := range state.Selectors {
			if !utf8.ValidString(selector) {
				return fmt.Errorf("assignment policy eligibility selector must be valid UTF-8")
			}
			if strings.TrimSpace(selector) == "" {
				return fmt.Errorf("assignment policy eligibility selector is empty")
			}
			if strings.TrimSpace(selector) != selector {
				return fmt.Errorf(
					"assignment policy eligibility selector %q is not canonical: surrounding whitespace is forbidden",
					selector,
				)
			}
			if len(selector) > cp.ComponentManagedMaximumPhysicalKey {
				return fmt.Errorf(
					"assignment policy eligibility selector exceeds %d bytes",
					cp.ComponentManagedMaximumPhysicalKey,
				)
			}
		}
	}
	return nil
}

type assignmentPolicySourcePlanner struct {
	store             cp.ComponentManagedReadStore
	targetsBySelector map[string]map[string]assignmentPolicySourceTarget
	selectorNames     map[string]string
	targetsByID       map[string]assignmentPolicySourceTarget
	selectorQueries   int
	selectorEdges     int
}

func newAssignmentPolicySourcePlanner(
	store cp.ComponentManagedReadStore,
) *assignmentPolicySourcePlanner {
	return &assignmentPolicySourcePlanner{
		store:             store,
		targetsBySelector: make(map[string]map[string]assignmentPolicySourceTarget),
		selectorNames:     make(map[string]string),
		targetsByID:       make(map[string]assignmentPolicySourceTarget),
	}
}

func (planner *assignmentPolicySourcePlanner) targetsForSelectors(
	ctx context.Context,
	selectors []string,
) (map[string]assignmentPolicySourceTarget, error) {
	targets := make(map[string]assignmentPolicySourceTarget)
	for _, selector := range selectors {
		selected, err := planner.targetsForSelector(ctx, selector)
		if err != nil {
			return nil, err
		}
		for id, target := range selected {
			targets[id] = target
		}
	}
	return targets, nil
}

func (planner *assignmentPolicySourcePlanner) targetsForSelector(
	ctx context.Context,
	selector string,
) (map[string]assignmentPolicySourceTarget, error) {
	identity := assignmentPolicyDatabaseIdentity(selector)
	if cached, ok := planner.targetsBySelector[identity]; ok {
		return cached, nil
	}
	planner.selectorQueries++
	if planner.selectorQueries > assignmentPolicyMaximumRows {
		return nil, fmt.Errorf(
			"assignment policy eligibility transition consults %d selector keys; maximum is %d",
			planner.selectorQueries,
			assignmentPolicyMaximumRows,
		)
	}
	remainingEdges := assignmentPolicyMaximumRows - planner.selectorEdges
	rows, exceeded, err := readAssignmentPolicyRowsUpTo(ctx, planner.store, cp.ComponentManagedRowQuery{
		Table:   schemacatalog.AssignmentPolicySelectorTable,
		Columns: []string{"ProfileName", "PermissionSetId"},
		Equal:   cp.ComponentRow{"ProfileName": selector},
		OrderBy: []cp.ComponentManagedOrder{
			{Column: "PermissionSetId", Direction: cp.ComponentManagedOrderAscending},
			{Column: "ProfileName", Direction: cp.ComponentManagedOrderAscending},
		},
		LogicalKeyColumns: []string{"PermissionSetId", "ProfileName"},
		MaximumRows:       assignmentPolicyMaximumRows,
	}, remainingEdges)
	if err != nil {
		return nil, err
	}
	planner.selectorEdges += len(rows)
	if exceeded {
		return nil, fmt.Errorf(
			"assignment policy eligibility transition expands to %d selector-target edges; maximum is %d",
			planner.selectorEdges,
			assignmentPolicyMaximumRows,
		)
	}
	targets := make(map[string]assignmentPolicySourceTarget, len(rows))
	for _, row := range rows {
		id, rowErr := assignmentPolicyRequiredString(row, "PermissionSetId")
		if rowErr != nil {
			return nil, rowErr
		}
		target, rowErr := planner.targetByID(ctx, id)
		if rowErr != nil {
			return nil, rowErr
		}
		targets[id] = target
	}
	planner.targetsBySelector[identity] = targets
	planner.selectorNames[identity] = selector
	return targets, nil
}

func (planner *assignmentPolicySourcePlanner) canonicalSelectorMapping() []assignmentPolicySourceSelector {
	identities := make([]string, 0, len(planner.targetsBySelector))
	for identity := range planner.targetsBySelector {
		identities = append(identities, identity)
	}
	sort.Strings(identities)
	out := make([]assignmentPolicySourceSelector, 0, len(identities))
	for _, identity := range identities {
		targets := planner.targetsBySelector[identity]
		permissionSetIDs := make([]string, 0, len(targets))
		for permissionSetID := range targets {
			permissionSetIDs = append(permissionSetIDs, permissionSetID)
		}
		sort.Strings(permissionSetIDs)
		out = append(out, assignmentPolicySourceSelector{
			Selector: planner.selectorNames[identity], PermissionSetIDs: permissionSetIDs,
		})
	}
	return out
}

func (planner *assignmentPolicySourcePlanner) targetByID(
	ctx context.Context,
	id string,
) (assignmentPolicySourceTarget, error) {
	identity := assignmentPolicyDatabaseIdentity(id)
	if cached, ok := planner.targetsByID[identity]; ok {
		return cached, nil
	}
	name, definitionOwnerPack, err := readAssignmentPolicySourceTargetRootByID(ctx, planner.store, id)
	if err != nil {
		return assignmentPolicySourceTarget{}, err
	}
	claims, err := planner.store.ReadRows(ctx, assignmentPolicyQuery(
		schemacatalog.AssignmentPolicyClaimTable,
		[]string{"PermissionSetId", "OwnerPack", "PolicyKey"},
		cp.ComponentRow{"PermissionSetId": id},
		1,
		nil,
	))
	if err != nil {
		return assignmentPolicySourceTarget{}, err
	}
	if len(claims) != 1 {
		return assignmentPolicySourceTarget{}, fmt.Errorf(
			"assignment policy selector target %s has no unique authority claim", id,
		)
	}
	ownerPack, err := assignmentPolicyRequiredString(claims[0], "OwnerPack")
	if err != nil {
		return assignmentPolicySourceTarget{}, err
	}
	policyKey, err := assignmentPolicyRequiredString(claims[0], "PolicyKey")
	if err != nil {
		return assignmentPolicySourceTarget{}, err
	}
	target := assignmentPolicySourceTarget{
		PermissionSetID:     id,
		PermissionSet:       name,
		DefinitionOwnerPack: definitionOwnerPack,
		OwnerPack:           ownerPack,
		PolicyKey:           policyKey,
	}
	planner.targetsByID[identity] = target
	return target, nil
}

func countAssignmentPolicySourceTargetAssignments(
	ctx context.Context,
	store cp.ComponentManagedReadStore,
	permissionSetID string,
	maximumRows int,
) (int, bool, error) {
	rows, exceeded, err := readAssignmentPolicyRowsUpTo(ctx, store, cp.ComponentManagedRowQuery{
		Table:             "PermissionSetAssignment",
		Columns:           []string{"Id"},
		Equal:             cp.ComponentRow{"PermissionSetId": permissionSetID},
		OrderBy:           []cp.ComponentManagedOrder{{Column: "Id", Direction: cp.ComponentManagedOrderAscending}},
		LogicalKeyColumns: []string{"Id"},
		MaximumRows:       assignmentPolicyMaximumRows,
	}, maximumRows)
	if err != nil {
		return 0, false, err
	}
	return len(rows), exceeded, nil
}

func readAssignmentPolicySourceTargetSelectors(
	ctx context.Context,
	store cp.ComponentManagedReadStore,
	permissionSetID string,
) ([]string, error) {
	selectors, exceeded, err := readAssignmentPolicySourceTargetSelectorsUpTo(
		ctx, store, permissionSetID, assignmentPolicyMaximumRows,
	)
	if err != nil {
		return nil, err
	}
	if exceeded {
		return nil, fmt.Errorf(
			"assignment policy source target %s has more than %d eligibility selectors",
			permissionSetID, assignmentPolicyMaximumRows,
		)
	}
	return selectors, nil
}

func readAssignmentPolicySourceTargetSelectorsUpTo(
	ctx context.Context,
	store cp.ComponentManagedReadStore,
	permissionSetID string,
	maximumRows int,
) ([]string, bool, error) {
	selectorRows, exceeded, err := readAssignmentPolicyRowsUpTo(
		ctx,
		store,
		cp.ComponentManagedRowQuery{
			Table:             schemacatalog.AssignmentPolicySelectorTable,
			Columns:           []string{"PermissionSetId", "ProfileName"},
			Equal:             cp.ComponentRow{"PermissionSetId": permissionSetID},
			OrderBy:           []cp.ComponentManagedOrder{{Column: "ProfileName", Direction: cp.ComponentManagedOrderAscending}},
			LogicalKeyColumns: []string{"ProfileName"},
			MaximumRows:       assignmentPolicyMaximumRows,
		},
		maximumRows,
	)
	if err != nil {
		return nil, false, err
	}
	selectors := make([]string, 0, len(selectorRows))
	for _, selectorRow := range selectorRows {
		actualPermissionSetID, rowErr := assignmentPolicyRequiredString(
			selectorRow, "PermissionSetId",
		)
		if rowErr != nil {
			return nil, false, rowErr
		}
		if actualPermissionSetID != permissionSetID {
			return nil, false, fmt.Errorf(
				"assignment policy target selector query returned another Permission Set",
			)
		}
		profileName, rowErr := assignmentPolicyRequiredString(selectorRow, "ProfileName")
		if rowErr != nil {
			return nil, false, rowErr
		}
		selectors = append(selectors, profileName)
	}
	return selectors, exceeded, nil
}

func lockAndValidateAssignmentPolicySourceSelectorMapping(
	ctx context.Context,
	store cp.ComponentManagedTransactionMutationStore,
	planned []assignmentPolicySourceSelector,
) error {
	for _, binding := range planned {
		rows, exceeded, err := lockAssignmentPolicyRowsUpTo(ctx, store, cp.ComponentManagedRowQuery{
			Table:   schemacatalog.AssignmentPolicySelectorTable,
			Columns: []string{"ProfileName", "PermissionSetId"},
			Equal:   cp.ComponentRow{"ProfileName": binding.Selector},
			OrderBy: []cp.ComponentManagedOrder{
				{Column: "PermissionSetId", Direction: cp.ComponentManagedOrderAscending},
				{Column: "ProfileName", Direction: cp.ComponentManagedOrderAscending},
			},
			LogicalKeyColumns: []string{"PermissionSetId", "ProfileName"},
			MaximumRows:       assignmentPolicyMaximumRows,
		}, len(binding.PermissionSetIDs))
		if err != nil {
			return err
		}
		if exceeded {
			return fmt.Errorf(
				"assignment policy source selector mapping changed for %q: planned %v, locked population exceeds %d",
				binding.Selector, binding.PermissionSetIDs, len(binding.PermissionSetIDs),
			)
		}
		actual := make([]string, 0, len(rows))
		seen := make(map[string]struct{}, len(rows))
		for _, row := range rows {
			selector, rowErr := assignmentPolicyRequiredString(row, "ProfileName")
			if rowErr != nil {
				return rowErr
			}
			if !assignmentPolicyDatabaseIdentityEqual(selector, binding.Selector) {
				return fmt.Errorf("assignment policy source selector mapping returned another selector")
			}
			permissionSetID, rowErr := assignmentPolicyRequiredString(row, "PermissionSetId")
			if rowErr != nil {
				return rowErr
			}
			if _, duplicate := seen[permissionSetID]; duplicate {
				return fmt.Errorf("assignment policy source selector mapping contains a duplicate target")
			}
			seen[permissionSetID] = struct{}{}
			actual = append(actual, permissionSetID)
		}
		sort.Strings(actual)
		if !slices.Equal(actual, binding.PermissionSetIDs) {
			return fmt.Errorf(
				"assignment policy source selector mapping changed for %q: planned %v, locked %v",
				binding.Selector, binding.PermissionSetIDs, actual,
			)
		}
	}
	return nil
}

func readAssignmentPolicySourceTargetRootByID(
	ctx context.Context,
	store cp.ComponentManagedReadStore,
	id string,
) (string, string, error) {
	rows, err := store.ReadRows(ctx, assignmentPolicyRootQuery(cp.ComponentRow{"Id": id}))
	if err != nil {
		return "", "", err
	}
	if len(rows) != 1 {
		return "", "", fmt.Errorf("assignment policy source target %s is missing", id)
	}
	actualID, name, ownerPack, err := assignmentPolicyDefinitionRootFromRow(rows[0])
	if err != nil {
		return "", "", fmt.Errorf("assignment policy source target %s: %w", id, err)
	}
	if actualID != id {
		return "", "", fmt.Errorf("assignment policy source target %s changed identity", id)
	}
	return name, ownerPack, nil
}

func decodeAssignmentPolicySourcePlan(plan cp.ComponentManagedPlan) (assignmentPolicySourcePlan, error) {
	var decoded assignmentPolicySourcePlan
	if err := json.Unmarshal(plan.OpaquePlan(), &decoded); err != nil {
		return assignmentPolicySourcePlan{}, err
	}
	if len(decoded.Mutations) == 0 {
		return assignmentPolicySourcePlan{}, fmt.Errorf("assignment policy source plan has no mutations")
	}
	for _, mutation := range decoded.Mutations {
		if err := validateAssignmentPolicySourceMutation(mutation); err != nil {
			return assignmentPolicySourcePlan{}, fmt.Errorf("assignment policy source plan: %w", err)
		}
	}
	if err := validateAssignmentPolicySourceSelectorMapping(
		decoded.Mutations, decoded.SelectorMapping,
	); err != nil {
		return assignmentPolicySourcePlan{}, fmt.Errorf("assignment policy source plan: %w", err)
	}
	return decoded, nil
}

func validateAssignmentPolicySourceSelectorMapping(
	mutations []AssignmentPolicyEligibilityMutation,
	planned []assignmentPolicySourceSelector,
) error {
	if len(planned) == 0 || len(planned) > assignmentPolicyMaximumRows {
		return fmt.Errorf(
			"selector mapping must contain 1..%d selector keys",
			assignmentPolicyMaximumRows,
		)
	}
	seen := make(map[string]struct{}, len(planned))
	previousIdentity := ""
	for index, binding := range planned {
		if _, err := assignmentPolicySelectorManagedTarget(binding.Selector); err != nil {
			return err
		}
		identity := assignmentPolicyDatabaseIdentity(binding.Selector)
		if index > 0 && identity <= previousIdentity {
			return fmt.Errorf("selector mapping is not canonical")
		}
		previousIdentity = identity
		seen[identity] = struct{}{}
		if len(binding.PermissionSetIDs) > assignmentPolicyMaximumRows ||
			!sort.StringsAreSorted(binding.PermissionSetIDs) {
			return fmt.Errorf("selector mapping targets are not canonical")
		}
		for targetIndex, permissionSetID := range binding.PermissionSetIDs {
			if strings.TrimSpace(permissionSetID) == "" {
				return fmt.Errorf("selector mapping contains an empty target")
			}
			if targetIndex > 0 && binding.PermissionSetIDs[targetIndex-1] == permissionSetID {
				return fmt.Errorf("selector mapping contains a duplicate target")
			}
		}
	}
	for _, mutation := range mutations {
		for _, state := range []AssignmentPolicyEligibilityState{mutation.Before, mutation.After} {
			for _, selector := range state.Selectors {
				if _, ok := seen[assignmentPolicyDatabaseIdentity(selector)]; !ok {
					return fmt.Errorf("selector mapping is missing an active selector")
				}
			}
		}
	}
	return nil
}

func canonicalAssignmentPolicySourceMutations(
	mutations []AssignmentPolicyEligibilityMutation,
) ([]AssignmentPolicyEligibilityMutation, error) {
	if len(mutations) > assignmentPolicyMaximumRows {
		return nil, fmt.Errorf(
			"assignment policy eligibility batch contains %d transitions; maximum is %d",
			len(mutations), assignmentPolicyMaximumRows,
		)
	}
	type mutationChain struct {
		sourceIdentity string
		edges          map[string]AssignmentPolicyEligibilityMutation
	}
	chains := make(map[string]*mutationChain, len(mutations))
	selectorReferences := 0
	inputBytes := 0
	for _, mutation := range mutations {
		if err := validateAssignmentPolicySourceMutation(mutation); err != nil {
			return nil, err
		}
		selectorReferences += len(mutation.Before.Selectors) + len(mutation.After.Selectors)
		if selectorReferences > 2*assignmentPolicyMaximumRows {
			return nil, fmt.Errorf(
				"assignment policy eligibility batch contains %d selector references; maximum is %d",
				selectorReferences, 2*assignmentPolicyMaximumRows,
			)
		}
		inputBytes += len(mutation.SourceIdentity)
		for _, selector := range mutation.Before.Selectors {
			inputBytes += len(selector)
		}
		for _, selector := range mutation.After.Selectors {
			inputBytes += len(selector)
		}
		if inputBytes > cp.ComponentManagedMaximumOpaquePlanBytes/2 {
			return nil, fmt.Errorf(
				"assignment policy eligibility batch input exceeds %d bytes",
				cp.ComponentManagedMaximumOpaquePlanBytes/2,
			)
		}
		mutation.Before = canonicalAssignmentPolicyEligibilityState(mutation.Before)
		mutation.After = canonicalAssignmentPolicyEligibilityState(mutation.After)
		identity := mutation.SourceIdentity
		chain := chains[identity]
		if chain == nil {
			chain = &mutationChain{
				sourceIdentity: mutation.SourceIdentity,
				edges:          make(map[string]AssignmentPolicyEligibilityMutation),
			}
			chains[identity] = chain
			if len(chains) > assignmentPolicyMaximumRows {
				return nil, fmt.Errorf(
					"assignment policy eligibility batch contains %d identities; maximum is %d",
					len(chains), assignmentPolicyMaximumRows,
				)
			}
		}
		edgeKey := assignmentPolicyEligibilityStateKey(mutation.Before) + "\x00" +
			assignmentPolicyEligibilityStateKey(mutation.After)
		if existing, duplicate := chain.edges[edgeKey]; duplicate {
			existing.ReconcileAfterSelectors = existing.ReconcileAfterSelectors ||
				mutation.ReconcileAfterSelectors
			chain.edges[edgeKey] = existing
			continue
		}
		chain.edges[edgeKey] = mutation
	}
	out := make([]AssignmentPolicyEligibilityMutation, 0, len(chains))
	for _, chain := range chains {
		mutation, include, err := composeAssignmentPolicyEligibilityMutationChain(
			chain.sourceIdentity,
			chain.edges,
		)
		if err != nil {
			return nil, err
		}
		if include {
			out = append(out, mutation)
		}
	}
	sort.Slice(out, func(i, j int) bool {
		return out[i].SourceIdentity < out[j].SourceIdentity
	})
	return out, nil
}

func composeAssignmentPolicyEligibilityMutationChain(
	sourceIdentity string,
	edges map[string]AssignmentPolicyEligibilityMutation,
) (AssignmentPolicyEligibilityMutation, bool, error) {
	selfEdges := make(map[string]AssignmentPolicyEligibilityMutation)
	nonSelf := make([]AssignmentPolicyEligibilityMutation, 0, len(edges))
	for _, edge := range edges {
		if assignmentPolicyEligibilityStatesEqual(edge.Before, edge.After) {
			selfEdges[assignmentPolicyEligibilityStateKey(edge.Before)] = edge
			continue
		}
		nonSelf = append(nonSelf, edge)
	}
	if len(selfEdges) > 1 {
		return AssignmentPolicyEligibilityMutation{}, false, fmt.Errorf(
			"assignment policy eligibility batch contains disconnected self transitions for %s",
			sourceIdentity,
		)
	}
	if len(nonSelf) == 0 {
		if len(selfEdges) == 0 {
			return AssignmentPolicyEligibilityMutation{}, false, nil
		}
		for _, self := range selfEdges {
			if !self.ReconcileAfterSelectors {
				return AssignmentPolicyEligibilityMutation{}, false, nil
			}
			self.SourceIdentity = sourceIdentity
			return self, true, nil
		}
	}

	outgoing := make(map[string]AssignmentPolicyEligibilityMutation, len(nonSelf))
	incoming := make(map[string]AssignmentPolicyEligibilityMutation, len(nonSelf))
	states := make(map[string]AssignmentPolicyEligibilityState, 2*len(nonSelf))
	for _, edge := range nonSelf {
		beforeKey := assignmentPolicyEligibilityStateKey(edge.Before)
		afterKey := assignmentPolicyEligibilityStateKey(edge.After)
		if existing, duplicate := outgoing[beforeKey]; duplicate &&
			assignmentPolicyEligibilityStateKey(existing.After) != afterKey {
			return AssignmentPolicyEligibilityMutation{}, false, fmt.Errorf(
				"assignment policy eligibility batch contains conflicting transitions for %s",
				sourceIdentity,
			)
		}
		if existing, duplicate := incoming[afterKey]; duplicate &&
			assignmentPolicyEligibilityStateKey(existing.Before) != beforeKey {
			return AssignmentPolicyEligibilityMutation{}, false, fmt.Errorf(
				"assignment policy eligibility batch contains conflicting transitions for %s",
				sourceIdentity,
			)
		}
		outgoing[beforeKey] = edge
		incoming[afterKey] = edge
		states[beforeKey] = edge.Before
		states[afterKey] = edge.After
	}
	starts := make([]string, 0, 1)
	for stateKey := range states {
		if _, hasOutgoing := outgoing[stateKey]; !hasOutgoing {
			continue
		}
		if _, hasIncoming := incoming[stateKey]; !hasIncoming {
			starts = append(starts, stateKey)
		}
	}
	if len(starts) != 1 {
		return AssignmentPolicyEligibilityMutation{}, false, fmt.Errorf(
			"assignment policy eligibility batch contains conflicting transitions for %s",
			sourceIdentity,
		)
	}
	start := starts[0]
	cursor := start
	visited := make(map[string]struct{}, len(nonSelf))
	for {
		edge, ok := outgoing[cursor]
		if !ok {
			break
		}
		edgeKey := assignmentPolicyEligibilityStateKey(edge.Before) + "\x00" +
			assignmentPolicyEligibilityStateKey(edge.After)
		if _, duplicate := visited[edgeKey]; duplicate {
			return AssignmentPolicyEligibilityMutation{}, false, fmt.Errorf(
				"assignment policy eligibility batch contains a transition cycle for %s",
				sourceIdentity,
			)
		}
		visited[edgeKey] = struct{}{}
		cursor = assignmentPolicyEligibilityStateKey(edge.After)
	}
	if len(visited) != len(nonSelf) {
		return AssignmentPolicyEligibilityMutation{}, false, fmt.Errorf(
			"assignment policy eligibility batch contains disconnected transitions for %s",
			sourceIdentity,
		)
	}
	reconcile := false
	for _, edge := range nonSelf {
		if !edge.ReconcileAfterSelectors {
			continue
		}
		if assignmentPolicyEligibilityStateKey(edge.After) != cursor {
			return AssignmentPolicyEligibilityMutation{}, false, fmt.Errorf(
				"assignment policy eligibility batch contains a non-terminal reconcile transition for %s",
				sourceIdentity,
			)
		}
		reconcile = true
	}
	for stateKey, self := range selfEdges {
		if _, connected := states[stateKey]; !connected {
			return AssignmentPolicyEligibilityMutation{}, false, fmt.Errorf(
				"assignment policy eligibility batch contains a disconnected self transition for %s",
				sourceIdentity,
			)
		}
		if self.ReconcileAfterSelectors {
			if stateKey != cursor {
				return AssignmentPolicyEligibilityMutation{}, false, fmt.Errorf(
					"assignment policy eligibility batch contains a non-terminal reconcile transition for %s",
					sourceIdentity,
				)
			}
			reconcile = true
		}
	}
	return AssignmentPolicyEligibilityMutation{
		SourceIdentity:          sourceIdentity,
		Before:                  states[start],
		After:                   states[cursor],
		ReconcileAfterSelectors: reconcile,
	}, true, nil
}

func canonicalAssignmentPolicyEligibilityState(
	state AssignmentPolicyEligibilityState,
) AssignmentPolicyEligibilityState {
	byIdentity := make(map[string]string, len(state.Selectors))
	for _, selector := range state.Selectors {
		identity := assignmentPolicyDatabaseIdentity(selector)
		if existing, ok := byIdentity[identity]; !ok || selector < existing {
			byIdentity[identity] = selector
		}
	}
	selectors := make([]string, 0, len(byIdentity))
	for _, selector := range byIdentity {
		selectors = append(selectors, selector)
	}
	sort.Slice(selectors, func(i, j int) bool {
		left := assignmentPolicyDatabaseIdentity(selectors[i])
		right := assignmentPolicyDatabaseIdentity(selectors[j])
		if left != right {
			return left < right
		}
		return selectors[i] < selectors[j]
	})
	return AssignmentPolicyEligibilityState{Selectors: selectors}
}

func assignmentPolicyEligibilityStateKey(state AssignmentPolicyEligibilityState) string {
	canonical := canonicalAssignmentPolicyEligibilityState(state)
	identities := make([]string, len(canonical.Selectors))
	for index, selector := range canonical.Selectors {
		identities[index] = assignmentPolicyDatabaseIdentity(selector)
	}
	encoded, _ := json.Marshal(identities)
	return string(encoded)
}

func assignmentPolicyEligibilityStatesEqual(
	left AssignmentPolicyEligibilityState,
	right AssignmentPolicyEligibilityState,
) bool {
	return assignmentPolicyEligibilityStateKey(left) == assignmentPolicyEligibilityStateKey(right)
}

func assignmentPolicySourceMutationIdentity(mutations []AssignmentPolicyEligibilityMutation) string {
	parts := make([]string, 0, len(mutations))
	for _, mutation := range mutations {
		parts = append(parts, mutation.SourceIdentity)
	}
	encoded, _ := json.Marshal(parts)
	return string(encoded)
}

func canonicalAssignmentPolicySourceTargets(input map[string]assignmentPolicySourceTarget) []assignmentPolicySourceTarget {
	out := make([]assignmentPolicySourceTarget, 0, len(input))
	for _, target := range input {
		out = append(out, target)
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].PermissionSetID != out[j].PermissionSetID {
			return out[i].PermissionSetID < out[j].PermissionSetID
		}
		if out[i].OwnerPack != out[j].OwnerPack {
			return out[i].OwnerPack < out[j].OwnerPack
		}
		return out[i].PolicyKey < out[j].PolicyKey
	})
	return out
}

func assignmentPolicySourceTargetUnion(groups ...[]assignmentPolicySourceTarget) []assignmentPolicySourceTarget {
	union := make(map[string]assignmentPolicySourceTarget)
	for _, group := range groups {
		for _, target := range group {
			union[target.PermissionSetID] = target
		}
	}
	return canonicalAssignmentPolicySourceTargets(union)
}

func assignmentPolicySourceTargetAssignmentRows(
	targets []assignmentPolicySourceTarget,
) map[string]int {
	out := make(map[string]int, len(targets))
	for _, target := range targets {
		out[target.PermissionSetID] = target.AssignmentRows
	}
	return out
}

func assignmentPolicySourceTargetEligibilityBudgetItems(
	targets []assignmentPolicySourceTarget,
) map[string]int {
	out := make(map[string]int, len(targets))
	for _, target := range targets {
		out[target.PermissionSetID] = target.EligibilityReadItems + target.EligibilitySubjects
	}
	return out
}

func reconcileAssignmentPolicySourceTargetAssignments(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	ownerPack, policyKey string,
	permissionSetID string,
	generation uint64,
	eligible []string,
	plannedAssignmentRows int,
) (int, error) {
	rows, exceeded, err := lockAssignmentPolicyRowsUpTo(ctx, store, cp.ComponentManagedRowQuery{
		Table:   "PermissionSetAssignment",
		Columns: []string{"Id", "PermissionSetId", "AssigneeId", "IsDeleted", "LastModifiedDate"},
		Equal:   cp.ComponentRow{"PermissionSetId": permissionSetID},
		OrderBy: []cp.ComponentManagedOrder{
			{Column: "PermissionSetId", Direction: cp.ComponentManagedOrderAscending},
			{Column: "AssigneeId", Direction: cp.ComponentManagedOrderAscending},
		},
		LogicalKeyColumns: []string{"PermissionSetId", "AssigneeId"},
		MaximumRows:       assignmentPolicyMaximumRows,
	}, plannedAssignmentRows)
	if err != nil {
		return 0, err
	}
	if exceeded {
		return 0, fmt.Errorf(
			"assignment policy source target %s assignment population grew from %d to at least %d after locked planning",
			permissionSetID, plannedAssignmentRows, len(rows),
		)
	}
	return reconcileAssignmentPolicyAssignmentRows(
		ctx,
		store,
		ownerPack,
		policyKey,
		permissionSetID,
		generation,
		false,
		nil,
		false,
		rows,
		eligible,
	)
}