package permissions

import (
	"bytes"
	"context"
	"crypto/sha256"
	"fmt"
	"math"
	"sort"
	"strconv"
	"strings"
	"time"

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

const (
	assignmentPolicyResolutionResolved         = "resolved"
	assignmentPolicyResolutionInvalidSelector  = "invalid_selector"
	assignmentPolicyResolutionUnsupportedDrift = "unsupported_definition_drift"
	assignmentPolicyResolutionRemoved          = "removed"
	assignmentPolicyMaximumRows                = 500
	assignmentPolicyMutationBytesPerRow        = 512
)

type assignmentPolicyRootLock struct {
	PermissionSetID string
	PermissionSet   string
	OwnerPack       string
}

type assignmentPolicyTargetSnapshot struct {
	PermissionSetID     string
	PermissionSet       string
	DefinitionOwnerPack string
	ClaimID             string
	OwnerPack           string
	PolicyKey           string
	Profiles            []string
	Eligible            []string
	Assignments         map[string]assignmentPolicyAssignmentRow
	State               *assignmentPolicyStateRow
	Resolution          string
}

type assignmentPolicyAssignmentRow struct {
	ID               string
	Assignee         string
	IsDeleted        bool
	LastModifiedDate any
}

// assignmentPolicyReleaseProof is transaction-local authority created only
// after the target-bound receipt, removed generation, and exact-empty target
// have been validated while the physical PermissionSet root is locked.
type assignmentPolicyReleaseProof struct {
	OwnerPack          string
	PolicyKey          string
	PermissionSetID    string
	ReleasedGeneration uint64
}

type assignmentPolicyStateRow struct {
	ID                string
	Suspended         bool
	WorkGeneration    uint64
	AppliedGeneration uint64
	ResolutionState   string
}

func assignmentPolicyReleaseUsage(snapshot assignmentPolicyTargetSnapshot) cp.ComponentManagedUsage {
	activeAssignments := 0
	for _, assignment := range snapshot.Assignments {
		if !assignment.IsDeleted {
			activeAssignments++
		}
	}
	deletedAssignments := len(snapshot.Assignments) - activeAssignments
	// Release first converges active rows to tombstones, then physically deletes
	// every receipt-proven policy tombstone. An active row can therefore require
	// four mutations (soft delete, receipt write, receipt delete, row delete); an
	// already-deleted owned row can require three (receipt rebind and both
	// deletes). Treat every deleted row as owned for the conservative bound.
	deltaRows := len(snapshot.Profiles) + 4*activeAssignments + 3*deletedAssignments + 3
	return cp.ComponentManagedUsage{
		CurrentItems:       len(snapshot.Profiles) + 3*len(snapshot.Assignments) + 3,
		DeltaRows:          deltaRows,
		MutationStatements: 4*activeAssignments + 3*deletedAssignments + 4,
		MutationBytes:      deltaRows * assignmentPolicyMutationBytesPerRow,
	}
}

func readAssignmentPolicyTarget(ctx context.Context, store cp.ComponentManagedReadStore, name string) (assignmentPolicyTargetSnapshot, error) {
	return readAssignmentPolicyTargetAllowingDeclaredUnavailable(ctx, store, name, false)
}

func readAssignmentPolicyTargetAllowingDeclaredUnavailable(
	ctx context.Context,
	store cp.ComponentManagedReadStore,
	name string,
	allowDeclaredUnavailable bool,
) (assignmentPolicyTargetSnapshot, error) {
	rows, err := store.ReadRows(ctx, assignmentPolicyRootQuery(cp.ComponentRow{"Name": name}))
	if err != nil {
		return assignmentPolicyTargetSnapshot{}, err
	}
	if len(rows) == 0 && allowDeclaredUnavailable {
		return assignmentPolicyTargetSnapshot{
			PermissionSet: name,
			Assignments:   map[string]assignmentPolicyAssignmentRow{},
		}, nil
	}
	if len(rows) != 1 {
		return assignmentPolicyTargetSnapshot{}, fmt.Errorf("assignment policy target %q is not one live Permission Set Definition", name)
	}
	id, actualName, definitionOwnerPack, err := assignmentPolicyDefinitionRootFromRow(rows[0])
	if err != nil {
		if allowDeclaredUnavailable && assignmentPolicyDeclaredDependencyIsInactive(rows[0], name) {
			return assignmentPolicyTargetSnapshot{
				PermissionSet: name,
				Assignments:   map[string]assignmentPolicyAssignmentRow{},
			}, nil
		}
		return assignmentPolicyTargetSnapshot{}, fmt.Errorf("assignment policy target %q: %w", name, err)
	}
	snapshot := assignmentPolicyTargetSnapshot{
		PermissionSetID: id, PermissionSet: actualName, DefinitionOwnerPack: definitionOwnerPack,
		Assignments: map[string]assignmentPolicyAssignmentRow{},
	}
	claimRows, err := store.ReadRows(ctx, assignmentPolicyQuery(
		schemacatalog.AssignmentPolicyClaimTable,
		[]string{"Id", "PermissionSetId", "OwnerPack", "PolicyKey"},
		cp.ComponentRow{"PermissionSetId": id}, 1, nil,
	))
	if err != nil {
		return assignmentPolicyTargetSnapshot{}, err
	}
	if len(claimRows) == 1 {
		snapshot.ClaimID, err = assignmentPolicyRequiredString(claimRows[0], "Id")
		if err != nil {
			return assignmentPolicyTargetSnapshot{}, err
		}
		snapshot.OwnerPack, err = assignmentPolicyRequiredString(claimRows[0], "OwnerPack")
		if err != nil {
			return assignmentPolicyTargetSnapshot{}, err
		}
		snapshot.PolicyKey, err = assignmentPolicyRequiredString(claimRows[0], "PolicyKey")
		if err != nil {
			return assignmentPolicyTargetSnapshot{}, err
		}
	}
	selectorRows, err := readAssignmentPolicyRows(ctx, store, cp.ComponentManagedRowQuery{
		Table:             schemacatalog.AssignmentPolicySelectorTable,
		Columns:           []string{"PermissionSetId", "ProfileName"},
		Equal:             cp.ComponentRow{"PermissionSetId": id},
		OrderBy:           []cp.ComponentManagedOrder{{Column: "ProfileName", Direction: cp.ComponentManagedOrderAscending}},
		LogicalKeyColumns: []string{"ProfileName"}, MaximumRows: assignmentPolicyMaximumRows,
	})
	if err != nil {
		return assignmentPolicyTargetSnapshot{}, err
	}
	for _, row := range selectorRows {
		profile, rowErr := assignmentPolicyRequiredString(row, "ProfileName")
		if rowErr != nil {
			return assignmentPolicyTargetSnapshot{}, rowErr
		}
		snapshot.Profiles = append(snapshot.Profiles, profile)
	}
	stateRows, err := store.ReadRows(ctx, assignmentPolicyQuery(
		schemacatalog.AssignmentPolicyTargetStateTable,
		[]string{
			"Id", "TargetDomain", "PhysicalKey", "PermissionSetId", "Suspended",
			"WorkGeneration", "AppliedGeneration", "ResolutionState",
		},
		cp.ComponentRow{"PermissionSetId": id}, 1, nil,
	))
	if err != nil {
		return assignmentPolicyTargetSnapshot{}, err
	}
	if len(stateRows) == 1 {
		state, rowErr := assignmentPolicyStateFromRow(stateRows[0])
		if rowErr != nil {
			return assignmentPolicyTargetSnapshot{}, rowErr
		}
		if rowErr := assignmentPolicyValidateTargetStateIdentity(stateRows[0], id, actualName); rowErr != nil {
			return assignmentPolicyTargetSnapshot{}, rowErr
		}
		snapshot.State = &state
		snapshot.Resolution = state.ResolutionState
	}
	assignmentRows, err := readAssignmentPolicyRows(ctx, store, cp.ComponentManagedRowQuery{
		Table: "PermissionSetAssignment", Columns: []string{"Id", "PermissionSetId", "AssigneeId", "IsDeleted", "LastModifiedDate"},
		Equal:             cp.ComponentRow{"PermissionSetId": id},
		OrderBy:           []cp.ComponentManagedOrder{{Column: "AssigneeId", Direction: cp.ComponentManagedOrderAscending}},
		LogicalKeyColumns: []string{"AssigneeId"}, MaximumRows: assignmentPolicyMaximumRows,
	})
	if err != nil {
		return assignmentPolicyTargetSnapshot{}, err
	}
	for _, row := range assignmentRows {
		assignment, rowErr := assignmentPolicyAssignmentFromRow(row)
		if rowErr != nil {
			return assignmentPolicyTargetSnapshot{}, rowErr
		}
		identity := assignmentPolicyDatabaseIdentity(assignment.Assignee)
		if _, duplicate := snapshot.Assignments[identity]; duplicate {
			return assignmentPolicyTargetSnapshot{}, fmt.Errorf("duplicate PermissionSetAssignment pair for %s", assignment.Assignee)
		}
		snapshot.Assignments[identity] = assignment
	}
	return snapshot, nil
}

func assignmentPolicyDeclaredDependencyIsInactive(row cp.ComponentRow, name string) bool {
	actualName, err := assignmentPolicyRequiredString(row, "Name")
	if err != nil || actualName != name {
		return false
	}
	active, err := assignmentPolicyRequiredBool(row, "IsActive")
	if err != nil {
		return false
	}
	deleted, err := assignmentPolicyRequiredBool(row, "IsDeleted")
	return err == nil && (!active || deleted)
}

func assignmentPolicyNegativeEdgeExists(
	ctx context.Context,
	store cp.ComponentManagedReadStore,
	relation schemacatalog.PermissionSetDefinitionRelation,
	targetID string,
) (bool, error) {
	paged, ok := store.(cp.ComponentManagedKeysetTransactionReadStore)
	if !ok {
		return false, fmt.Errorf("assignment policy requires a bounded negative-edge existence read")
	}
	page, err := paged.ReadRowsKeysetPage(ctx, cp.ComponentManagedKeysetRowQuery{
		Query: cp.ComponentManagedRowQuery{
			Table:   relation.TableName,
			Columns: []string{"Id", relation.ParentField},
			Equal:   cp.ComponentRow{relation.ParentField: targetID},
			OrderBy: []cp.ComponentManagedOrder{{
				Column: "Id", Direction: cp.ComponentManagedOrderAscending,
			}},
			LogicalKeyColumns: []string{"Id"},
			MaximumRows:       1,
		},
	})
	if err != nil {
		return false, err
	}
	if len(page.Rows) == 0 && !page.Exhausted {
		return false, fmt.Errorf("assignment policy negative-edge existence read did not advance")
	}
	return len(page.Rows) != 0, nil
}

func assignmentPolicyNegativeEdges() ([]schemacatalog.PermissionSetDefinitionRelation, error) {
	inventory, err := schemacatalog.ManagedPermissionSetDefinitionInventory()
	if err != nil {
		return nil, fmt.Errorf("assignment policy Definition inventory: %w", err)
	}
	var relations []schemacatalog.PermissionSetDefinitionRelation
	for _, relation := range inventory.Relations() {
		if relation.Role != schemacatalog.PermissionSetDefinitionRoleNegativeEdge ||
			relation.TableName == schemacatalog.AssignmentPolicyClaimTable {
			continue
		}
		relations = append(relations, relation)
	}
	if len(relations) == 0 {
		return nil, fmt.Errorf("assignment policy Definition inventory has no negative authorization edges")
	}
	return relations, nil
}

func reconcileAssignmentPolicySelectors(ctx context.Context, store cp.ComponentManagedMutationStore, targetID string, desired []string) (int, error) {
	rows, err := lockAssignmentPolicyRows(ctx, store, cp.ComponentManagedRowQuery{
		Table:   schemacatalog.AssignmentPolicySelectorTable,
		Columns: []string{"Id", "PermissionSetId", "ProfileName"},
		Equal:   cp.ComponentRow{"PermissionSetId": targetID},
		OrderBy: []cp.ComponentManagedOrder{
			{Column: "PermissionSetId", Direction: cp.ComponentManagedOrderAscending},
			{Column: "ProfileName", Direction: cp.ComponentManagedOrderAscending},
		},
		LogicalKeyColumns: []string{"PermissionSetId", "ProfileName"},
		MaximumRows:       assignmentPolicyMaximumRows,
	})
	if err != nil {
		return 0, err
	}
	type currentSelector struct {
		id   string
		name string
	}
	current := make(map[string]currentSelector, len(rows))
	for _, row := range rows {
		id, rowErr := assignmentPolicyRequiredString(row, "Id")
		if rowErr != nil {
			return 0, rowErr
		}
		name, rowErr := assignmentPolicyRequiredString(row, "ProfileName")
		if rowErr != nil {
			return 0, rowErr
		}
		identity := assignmentPolicyDatabaseIdentity(name)
		if previous, duplicate := current[identity]; duplicate {
			return 0, fmt.Errorf(
				"assignment policy selectors %q and %q collide under database collation",
				previous.name, name,
			)
		}
		current[identity] = currentSelector{id: id, name: name}
	}
	desiredSet := make(map[string]string, len(desired))
	for _, name := range desired {
		identity := assignmentPolicyDatabaseIdentity(name)
		if previous, duplicate := desiredSet[identity]; duplicate {
			return 0, fmt.Errorf(
				"desired assignment policy selectors %q and %q collide under database collation",
				previous, name,
			)
		}
		desiredSet[identity] = name
	}
	changed := 0
	for key, selector := range current {
		desiredName, keep := desiredSet[key]
		if keep && selector.name == desiredName {
			continue
		}
		if keep {
			affected, updateErr := store.UpdateRows(ctx, cp.ComponentManagedUpdate{
				Table:                   schemacatalog.AssignmentPolicySelectorTable,
				Match:                   cp.ComponentRow{"Id": selector.id},
				Values:                  cp.ComponentRow{"ProfileName": desiredName},
				CurrentTimestampColumns: []string{"LastModifiedDate"}, MaximumAffectedRows: 1,
			})
			if err := assignmentPolicyRequireExactWrite(
				"update", schemacatalog.AssignmentPolicySelectorTable, affected, updateErr,
			); err != nil {
				return 0, err
			}
			changed++
			continue
		}
		affected, deleteErr := store.DeleteRows(ctx, cp.ComponentManagedDelete{
			Table: schemacatalog.AssignmentPolicySelectorTable,
			Match: cp.ComponentRow{"Id": selector.id}, MaximumAffectedRows: 1,
		})
		if err := assignmentPolicyRequireExactWrite(
			"delete", schemacatalog.AssignmentPolicySelectorTable, affected, deleteErr,
		); err != nil {
			return 0, err
		}
		changed++
	}
	for key, name := range desiredSet {
		if _, exists := current[key]; exists {
			continue
		}
		id, idErr := store.NewID()
		if idErr != nil {
			return 0, idErr
		}
		affected, insertErr := store.InsertRows(ctx, cp.ComponentManagedInsertBatch{
			Table: schemacatalog.AssignmentPolicySelectorTable,
			Rows: []cp.ComponentManagedInsertRow{{
				LogicalKey: cp.ComponentRow{"PermissionSetId": targetID, "ProfileName": name},
				Values:     cp.ComponentRow{"Id": id, "PermissionSetId": targetID, "ProfileName": name},
			}},
			CurrentTimestampColumns: []string{"CreatedDate", "LastModifiedDate"},
		})
		if err := assignmentPolicyRequireExactWrite(
			"insert", schemacatalog.AssignmentPolicySelectorTable, affected, insertErr,
		); err != nil {
			return 0, err
		}
		changed++
	}
	return changed, nil
}

func ensureAssignmentPolicyState(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	targetID, name, resolution string,
) (assignmentPolicyStateRow, int, error) {
	rows, err := store.LockRows(ctx, assignmentPolicyQuery(
		schemacatalog.AssignmentPolicyTargetStateTable,
		[]string{
			"Id", "TargetDomain", "PhysicalKey", "PermissionSetId", "Suspended",
			"WorkGeneration", "AppliedGeneration", "ResolutionState",
		},
		cp.ComponentRow{"PermissionSetId": targetID}, 1, []string{"PermissionSetId"},
	))
	if err != nil {
		return assignmentPolicyStateRow{}, 0, err
	}
	if len(rows) == 1 {
		if err := assignmentPolicyValidateTargetStateIdentity(rows[0], targetID, name); err != nil {
			return assignmentPolicyStateRow{}, 0, err
		}
		state, rowErr := assignmentPolicyStateFromRow(rows[0])
		return state, 0, rowErr
	}
	id, err := store.NewID()
	if err != nil {
		return assignmentPolicyStateRow{}, 0, err
	}
	affected, err := store.InsertRows(ctx, cp.ComponentManagedInsertBatch{
		Table: schemacatalog.AssignmentPolicyTargetStateTable,
		Rows: []cp.ComponentManagedInsertRow{{
			LogicalKey: cp.ComponentRow{"TargetDomain": permissionSetDefinitionManagedTargetDomain, "PhysicalKey": []byte(strings.ToLower(name))},
			Values: cp.ComponentRow{
				"Id": id, "TargetDomain": permissionSetDefinitionManagedTargetDomain, "PhysicalKey": []byte(strings.ToLower(name)),
				"PermissionSetId": targetID, "Suspended": false, "SuspensionActor": nil,
				"SuspensionReason": nil, "SuspendedAt": nil, "WorkGeneration": uint64(1),
				"AppliedGeneration": uint64(0), "NextAttemptAt": nil, "RetryCount": uint64(0),
				"LastFailureCode": nil, "LastFailureDigest": nil, "ResolutionState": resolution,
				"ResolutionDigest": assignmentPolicyDigest(resolution),
			},
		}},
		CurrentTimestampColumns: []string{"CreatedDate", "LastModifiedDate"},
	})
	if writeErr := assignmentPolicyRequireExactWrite(
		"insert", schemacatalog.AssignmentPolicyTargetStateTable, affected, err,
	); writeErr != nil {
		return assignmentPolicyStateRow{}, 0, writeErr
	}
	return assignmentPolicyStateRow{
		ID: id, WorkGeneration: 1, AppliedGeneration: 0, ResolutionState: resolution,
	}, 1, nil
}

func reconcileAssignmentPolicyAssignments(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	ownerPack, policyKey string,
	targetID string,
	generation uint64,
	requireReleasedProof bool,
	releaseProof *assignmentPolicyReleaseProof,
	rebindReleasedTombstones bool,
	desired []string,
) (int, error) {
	rows, err := lockAssignmentPolicyRows(ctx, store, cp.ComponentManagedRowQuery{
		Table:   "PermissionSetAssignment",
		Columns: []string{"Id", "PermissionSetId", "AssigneeId", "IsDeleted", "LastModifiedDate"},
		Equal:   cp.ComponentRow{"PermissionSetId": targetID},
		OrderBy: []cp.ComponentManagedOrder{
			{Column: "PermissionSetId", Direction: cp.ComponentManagedOrderAscending},
			{Column: "AssigneeId", Direction: cp.ComponentManagedOrderAscending},
		},
		LogicalKeyColumns: []string{"PermissionSetId", "AssigneeId"},
		MaximumRows:       assignmentPolicyMaximumRows,
	})
	if err != nil {
		return 0, err
	}
	return reconcileAssignmentPolicyAssignmentRows(
		ctx,
		store,
		ownerPack,
		policyKey,
		targetID,
		generation,
		requireReleasedProof,
		releaseProof,
		rebindReleasedTombstones,
		rows,
		desired,
	)
}

// reconcileAssignmentPolicyAssignmentRows applies an exact target projection
// to a population that its caller has already locked. Keeping the row-level
// reconciler separate lets target-started operation providers bound and prove
// the complete assignment population before the first mutation without taking
// the same lock set twice.
func reconcileAssignmentPolicyAssignmentRows(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	ownerPack, policyKey string,
	targetID string,
	generation uint64,
	requireReleasedProof bool,
	releaseProof *assignmentPolicyReleaseProof,
	rebindReleasedTombstones bool,
	rows []cp.ComponentRow,
	desired []string,
) (int, error) {
	current := make(map[string]assignmentPolicyAssignmentRow, len(rows))
	for _, row := range rows {
		assignment, rowErr := assignmentPolicyAssignmentFromRow(row)
		if rowErr != nil {
			return 0, rowErr
		}
		key := assignmentPolicyDatabaseIdentity(assignment.Assignee)
		if _, duplicate := current[key]; duplicate {
			return 0, fmt.Errorf("duplicate PermissionSetAssignment pair for %s", assignment.Assignee)
		}
		current[key] = assignment
	}
	desiredSet := make(map[string]string, len(desired))
	for _, id := range desired {
		desiredSet[assignmentPolicyDatabaseIdentity(id)] = id
	}
	// One exact-empty proof authorizes every exact tombstone from the same
	// source/target/generation. Re-reading exact emptiness after the first
	// reactivation would incorrectly reject the remaining tombstones.
	validatedReleaseProofs := make(map[string]uint64)
	if releaseProof != nil {
		validatedReleaseProofs[assignmentPolicyReleaseProofKey(
			releaseProof.OwnerPack, releaseProof.PolicyKey, releaseProof.PermissionSetID,
		)] = releaseProof.ReleasedGeneration
	}
	changed := 0
	for key, row := range current {
		_, want := desiredSet[key]
		switch {
		case row.IsDeleted && !want && rebindReleasedTombstones:
			receiptChanged, receiptErr := rebindAssignmentPolicyTombstoneReceiptForRelease(
				ctx, store, row, targetID, ownerPack, policyKey, generation,
			)
			if receiptErr != nil {
				return 0, receiptErr
			}
			changed += receiptChanged
		case row.IsDeleted && want:
			receiptChanged, receiptErr := consumeAssignmentPolicyTombstoneReceipt(
				ctx, store, row.ID, targetID, row.Assignee, ownerPack, policyKey,
				generation, row.LastModifiedDate, requireReleasedProof, validatedReleaseProofs,
			)
			if receiptErr != nil {
				return 0, receiptErr
			}
			affected, updateErr := store.UpdateRows(ctx, cp.ComponentManagedUpdate{
				Table: "PermissionSetAssignment", Match: cp.ComponentRow{"Id": row.ID},
				Values:                  cp.ComponentRow{"IsDeleted": false},
				CurrentTimestampColumns: []string{"LastModifiedDate"}, MaximumAffectedRows: 1,
			})
			if err := assignmentPolicyRequireExactWrite(
				"update", "PermissionSetAssignment", affected, updateErr,
			); err != nil {
				return 0, err
			}
			changed += receiptChanged + 1
		case !row.IsDeleted && !want:
			affected, updateErr := store.UpdateRows(ctx, cp.ComponentManagedUpdate{
				Table: "PermissionSetAssignment", Match: cp.ComponentRow{"Id": row.ID},
				Values:                  cp.ComponentRow{"IsDeleted": true},
				CurrentTimestampColumns: []string{"LastModifiedDate"}, MaximumAffectedRows: 1,
			})
			if err := assignmentPolicyRequireExactWrite(
				"update", "PermissionSetAssignment", affected, updateErr,
			); err != nil {
				return 0, err
			}
			tombstonedAt, eventErr := readAssignmentPolicyTombstoneEvent(
				ctx, store, row.ID, targetID, row.Assignee,
			)
			if eventErr != nil {
				return 0, eventErr
			}
			receiptChanged, receiptErr := upsertAssignmentPolicyTombstoneReceipt(
				ctx, store, row.ID, targetID, row.Assignee, ownerPack, policyKey,
				generation, tombstonedAt,
			)
			if receiptErr != nil {
				return 0, receiptErr
			}
			changed += receiptChanged + 1
		}
		delete(desiredSet, key)
	}
	for _, assignee := range desiredSet {
		id, idErr := store.NewID()
		if idErr != nil {
			return 0, idErr
		}
		affected, insertErr := store.InsertRows(ctx, cp.ComponentManagedInsertBatch{
			Table: "PermissionSetAssignment",
			Rows: []cp.ComponentManagedInsertRow{{
				LogicalKey: cp.ComponentRow{"PermissionSetId": targetID, "AssigneeId": assignee},
				Values: cp.ComponentRow{
					"Id": id, "PermissionSetId": targetID, "AssigneeId": assignee, "IsDeleted": false,
				},
			}},
			CurrentTimestampColumns: []string{"CreatedDate", "LastModifiedDate"},
		})
		if err := assignmentPolicyRequireExactWrite(
			"insert", "PermissionSetAssignment", affected, insertErr,
		); err != nil {
			return 0, err
		}
		changed++
	}
	return changed, nil
}

// rebindAssignmentPolicyTombstoneReceiptForRelease advances provenance owned
// by the releasing policy to the exact release generation without changing
// the underlying deletion event. Release finalization then uses that exact
// provenance to distinguish policy-owned tombstones from unrelated history.
func rebindAssignmentPolicyTombstoneReceiptForRelease(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	assignment assignmentPolicyAssignmentRow,
	permissionSetID, ownerPack, policyKey string,
	generation uint64,
) (int, error) {
	rows, err := store.LockRows(ctx, assignmentPolicyQuery(
		schemacatalog.AssignmentPolicyTombstoneReceiptTable,
		[]string{
			"Id", "AssignmentId", "PermissionSetId", "AssigneeId", "OwnerPack", "PolicyKey",
			"TombstonedGeneration", "TombstonedAt",
		},
		cp.ComponentRow{"AssignmentId": assignment.ID},
		1,
		[]string{"AssignmentId"},
	))
	if err != nil {
		return 0, err
	}
	if len(rows) == 0 {
		return 0, nil
	}
	receiptOwnerPack, err := assignmentPolicyRequiredString(rows[0], "OwnerPack")
	if err != nil {
		return 0, err
	}
	receiptPolicyKey, err := assignmentPolicyRequiredString(rows[0], "PolicyKey")
	if err != nil {
		return 0, err
	}
	if !assignmentPolicySourceIdentityEqual(
		receiptOwnerPack, receiptPolicyKey, ownerPack, policyKey,
	) {
		return 0, nil
	}
	receiptID, err := assignmentPolicyValidateTombstoneReceipt(
		rows[0], assignment.ID, permissionSetID, assignment.Assignee, ownerPack, policyKey,
	)
	if err != nil {
		return 0, err
	}
	receiptGeneration, err := assignmentPolicyRequiredUint64(rows[0], "TombstonedGeneration")
	if err != nil {
		return 0, err
	}
	if receiptGeneration == 0 || receiptGeneration >= generation {
		return 0, fmt.Errorf(
			"assignment policy tombstone receipt %s generation %d is not before release generation %d",
			receiptID, receiptGeneration, generation,
		)
	}
	tombstonedAt, err := assignmentPolicyRequiredTimestamp(rows[0], "TombstonedAt")
	if err != nil {
		return 0, err
	}
	assignmentEvent, err := assignmentPolicyRequiredTimestamp(
		cp.ComponentRow{"LastModifiedDate": assignment.LastModifiedDate}, "LastModifiedDate",
	)
	if err != nil {
		return 0, err
	}
	if !assignmentPolicyTimestampsEqual(tombstonedAt, assignmentEvent) {
		return 0, fmt.Errorf(
			"assignment policy tombstone receipt %s does not match the current deletion event",
			receiptID,
		)
	}
	affected, err := store.UpdateRows(ctx, cp.ComponentManagedUpdate{
		Table: schemacatalog.AssignmentPolicyTombstoneReceiptTable,
		Match: cp.ComponentRow{"Id": receiptID},
		Values: cp.ComponentRow{
			"TombstonedGeneration": generation,
		},
		CurrentTimestampColumns: []string{"LastModifiedDate"},
		MaximumAffectedRows:     1,
	})
	if err := assignmentPolicyRequireExactWrite(
		"update", schemacatalog.AssignmentPolicyTombstoneReceiptTable, affected, err,
	); err != nil {
		return 0, err
	}
	return 1, nil
}

// finalizeReleasedAssignmentPolicyTombstones permanently removes only deleted
// assignment identities whose receipt proves that the releasing policy owns
// the exact target, source, generation, assignee, and deletion event. A policy
// suspension never calls this function, so resume can still reactivate the
// original identity. After authority is released, a later policy grant creates
// a new identity instead of depending on an ordinary-create restore path.
func finalizeReleasedAssignmentPolicyTombstones(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	ownerPack, policyKey, permissionSetID string,
	releaseGeneration uint64,
) (int, error) {
	assignments, err := lockAssignmentPolicyRows(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,
	})
	if err != nil {
		return 0, err
	}
	byID := make(map[string]assignmentPolicyAssignmentRow, len(assignments))
	for _, row := range assignments {
		assignment, rowErr := assignmentPolicyAssignmentFromRow(row)
		if rowErr != nil {
			return 0, rowErr
		}
		byID[assignment.ID] = assignment
	}
	receipts, err := lockAssignmentPolicyRows(ctx, store, cp.ComponentManagedRowQuery{
		Table: schemacatalog.AssignmentPolicyTombstoneReceiptTable,
		Columns: []string{
			"Id", "AssignmentId", "PermissionSetId", "AssigneeId", "OwnerPack", "PolicyKey",
			"TombstonedGeneration", "TombstonedAt",
		},
		Equal:             cp.ComponentRow{"PermissionSetId": permissionSetID},
		OrderBy:           []cp.ComponentManagedOrder{{Column: "AssignmentId", Direction: cp.ComponentManagedOrderAscending}},
		LogicalKeyColumns: []string{"AssignmentId"},
		MaximumRows:       assignmentPolicyMaximumRows,
	})
	if err != nil {
		return 0, err
	}
	changed := 0
	for _, receipt := range receipts {
		receiptOwnerPack, rowErr := assignmentPolicyRequiredString(receipt, "OwnerPack")
		if rowErr != nil {
			return 0, rowErr
		}
		receiptPolicyKey, rowErr := assignmentPolicyRequiredString(receipt, "PolicyKey")
		if rowErr != nil {
			return 0, rowErr
		}
		if !assignmentPolicySourceIdentityEqual(
			receiptOwnerPack, receiptPolicyKey, ownerPack, policyKey,
		) {
			continue
		}
		assignmentID, rowErr := assignmentPolicyRequiredString(receipt, "AssignmentId")
		if rowErr != nil {
			return 0, rowErr
		}
		assignment, exists := byID[assignmentID]
		if !exists {
			return 0, fmt.Errorf(
				"assignment policy tombstone receipt for %s has no target assignment",
				assignmentID,
			)
		}
		if !assignment.IsDeleted {
			return 0, fmt.Errorf(
				"assignment policy tombstone receipt for %s names an active assignment",
				assignmentID,
			)
		}
		receiptID, rowErr := assignmentPolicyValidateTombstoneReceipt(
			receipt,
			assignment.ID,
			permissionSetID,
			assignment.Assignee,
			ownerPack,
			policyKey,
		)
		if rowErr != nil {
			return 0, rowErr
		}
		generation, rowErr := assignmentPolicyRequiredUint64(receipt, "TombstonedGeneration")
		if rowErr != nil {
			return 0, rowErr
		}
		if generation != releaseGeneration {
			return 0, fmt.Errorf(
				"assignment policy tombstone receipt %s generation %d differs from release generation %d",
				receiptID, generation, releaseGeneration,
			)
		}
		tombstonedAt, rowErr := assignmentPolicyRequiredTimestamp(receipt, "TombstonedAt")
		if rowErr != nil {
			return 0, rowErr
		}
		assignmentEvent, rowErr := assignmentPolicyRequiredTimestamp(
			cp.ComponentRow{"LastModifiedDate": assignment.LastModifiedDate}, "LastModifiedDate",
		)
		if rowErr != nil {
			return 0, rowErr
		}
		if !assignmentPolicyTimestampsEqual(tombstonedAt, assignmentEvent) {
			return 0, fmt.Errorf(
				"assignment policy tombstone receipt %s does not match the current deletion event",
				receiptID,
			)
		}
		deleted, deleteErr := store.DeleteRows(ctx, cp.ComponentManagedDelete{
			Table: schemacatalog.AssignmentPolicyTombstoneReceiptTable,
			Match: cp.ComponentRow{"Id": receiptID}, MaximumAffectedRows: 1,
		})
		if err := assignmentPolicyRequireExactWrite(
			"delete", schemacatalog.AssignmentPolicyTombstoneReceiptTable, deleted, deleteErr,
		); err != nil {
			return 0, err
		}
		deleted, deleteErr = store.DeleteRows(ctx, cp.ComponentManagedDelete{
			Table: "PermissionSetAssignment",
			Match: cp.ComponentRow{"Id": assignment.ID}, MaximumAffectedRows: 1,
		})
		if err := assignmentPolicyRequireExactWrite(
			"delete", "PermissionSetAssignment", deleted, deleteErr,
		); err != nil {
			return 0, err
		}
		changed += 2
	}
	return changed, nil
}

func readAssignmentPolicyTombstoneEvent(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	assignmentID, permissionSetID, assigneeID string,
) (any, error) {
	rows, err := store.LockRows(ctx, assignmentPolicyQuery(
		"PermissionSetAssignment",
		[]string{"Id", "PermissionSetId", "AssigneeId", "IsDeleted", "LastModifiedDate"},
		cp.ComponentRow{"Id": assignmentID},
		1,
		[]string{"Id"},
	))
	if err != nil {
		return nil, err
	}
	if len(rows) != 1 {
		return nil, fmt.Errorf("assignment policy tombstoned assignment %s is missing", assignmentID)
	}
	actualID, err := assignmentPolicyRequiredString(rows[0], "Id")
	if err != nil {
		return nil, err
	}
	actualPermissionSetID, err := assignmentPolicyRequiredString(rows[0], "PermissionSetId")
	if err != nil {
		return nil, err
	}
	actualAssigneeID, err := assignmentPolicyRequiredString(rows[0], "AssigneeId")
	if err != nil {
		return nil, err
	}
	deleted, err := assignmentPolicyRequiredBool(rows[0], "IsDeleted")
	if err != nil {
		return nil, err
	}
	if actualID != assignmentID || actualPermissionSetID != permissionSetID ||
		actualAssigneeID != assigneeID || !deleted {
		return nil, fmt.Errorf("assignment policy tombstone event identity changed for assignment %s", assignmentID)
	}
	return assignmentPolicyRequiredTimestamp(rows[0], "LastModifiedDate")
}

func upsertAssignmentPolicyTombstoneReceipt(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	assignmentID, permissionSetID, assigneeID, ownerPack, policyKey string,
	generation uint64,
	tombstonedAt any,
) (int, error) {
	rows, err := store.LockRows(ctx, assignmentPolicyQuery(
		schemacatalog.AssignmentPolicyTombstoneReceiptTable,
		[]string{"Id", "AssignmentId", "PermissionSetId", "AssigneeId", "OwnerPack", "PolicyKey"},
		cp.ComponentRow{"AssignmentId": assignmentID},
		1,
		[]string{"AssignmentId"},
	))
	if err != nil {
		return 0, err
	}
	values := cp.ComponentRow{
		"AssignmentId": assignmentID, "PermissionSetId": permissionSetID,
		"AssigneeId": assigneeID, "OwnerPack": ownerPack, "PolicyKey": policyKey,
		"TombstonedGeneration": generation, "TombstonedAt": tombstonedAt,
	}
	if len(rows) == 1 {
		receiptID, rowErr := assignmentPolicyValidateTombstoneReceipt(
			rows[0], assignmentID, permissionSetID, assigneeID, ownerPack, policyKey,
		)
		if rowErr != nil {
			return 0, rowErr
		}
		affected, updateErr := store.UpdateRows(ctx, cp.ComponentManagedUpdate{
			Table: schemacatalog.AssignmentPolicyTombstoneReceiptTable,
			Match: cp.ComponentRow{"Id": receiptID}, Values: values,
			CurrentTimestampColumns: []string{"LastModifiedDate"},
			MaximumAffectedRows:     1,
		})
		if err := assignmentPolicyRequireExactWrite(
			"update", schemacatalog.AssignmentPolicyTombstoneReceiptTable, affected, updateErr,
		); err != nil {
			return 0, err
		}
		return 1, nil
	}
	receiptID, err := store.NewID()
	if err != nil {
		return 0, err
	}
	values["Id"] = receiptID
	affected, err := store.InsertRows(ctx, cp.ComponentManagedInsertBatch{
		Table: schemacatalog.AssignmentPolicyTombstoneReceiptTable,
		Rows: []cp.ComponentManagedInsertRow{{
			LogicalKey: cp.ComponentRow{"AssignmentId": assignmentID}, Values: values,
		}},
		CurrentTimestampColumns: []string{"CreatedDate", "LastModifiedDate"},
	})
	if err := assignmentPolicyRequireExactWrite(
		"insert", schemacatalog.AssignmentPolicyTombstoneReceiptTable, affected, err,
	); err != nil {
		return 0, err
	}
	return 1, nil
}

func consumeAssignmentPolicyTombstoneReceipt(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	assignmentID, permissionSetID, assigneeID, ownerPack, policyKey string,
	currentGeneration uint64,
	assignmentLastModifiedDate any,
	requireReleasedProof bool,
	validatedReleaseProofs map[string]uint64,
) (int, error) {
	rows, err := store.LockRows(ctx, assignmentPolicyQuery(
		schemacatalog.AssignmentPolicyTombstoneReceiptTable,
		[]string{
			"Id", "AssignmentId", "PermissionSetId", "AssigneeId", "OwnerPack", "PolicyKey",
			"TombstonedGeneration", "TombstonedAt",
		},
		cp.ComponentRow{"AssignmentId": assignmentID},
		1,
		[]string{"AssignmentId"},
	))
	if err != nil {
		return 0, err
	}
	if len(rows) != 1 {
		return 0, fmt.Errorf(
			"PermissionSetAssignment %s is deleted without an exact assignment policy tombstone receipt",
			assignmentID,
		)
	}
	receiptOwnerPack, err := assignmentPolicyRequiredString(rows[0], "OwnerPack")
	if err != nil {
		return 0, err
	}
	receiptPolicyKey, err := assignmentPolicyRequiredString(rows[0], "PolicyKey")
	if err != nil {
		return 0, err
	}
	receiptID, err := assignmentPolicyValidateTombstoneReceipt(
		rows[0], assignmentID, permissionSetID, assigneeID, receiptOwnerPack, receiptPolicyKey,
	)
	if err != nil {
		return 0, err
	}
	receiptGeneration, err := assignmentPolicyRequiredUint64(rows[0], "TombstonedGeneration")
	if err != nil {
		return 0, err
	}
	if receiptGeneration == 0 || receiptGeneration >= currentGeneration {
		return 0, fmt.Errorf(
			"assignment policy tombstone receipt %s generation %d is not before current generation %d",
			receiptID, receiptGeneration, currentGeneration,
		)
	}
	tombstonedAt, err := assignmentPolicyRequiredTimestamp(rows[0], "TombstonedAt")
	if err != nil {
		return 0, err
	}
	assignmentEvent, err := assignmentPolicyRequiredTimestamp(
		cp.ComponentRow{"LastModifiedDate": assignmentLastModifiedDate}, "LastModifiedDate",
	)
	if err != nil {
		return 0, err
	}
	if !assignmentPolicyTimestampsEqual(tombstonedAt, assignmentEvent) {
		return 0, fmt.Errorf(
			"assignment policy tombstone receipt %s does not match the current deletion event",
			receiptID,
		)
	}
	sameSource := assignmentPolicySourceIdentityEqual(
		receiptOwnerPack, receiptPolicyKey, ownerPack, policyKey,
	)
	if requireReleasedProof || !sameSource {
		proofKey := assignmentPolicyReleaseProofKey(
			receiptOwnerPack, receiptPolicyKey, permissionSetID,
		)
		validatedGeneration, validated := validatedReleaseProofs[proofKey]
		if validated && validatedGeneration != receiptGeneration {
			return 0, fmt.Errorf("assignment policy release receipt generation differs from tombstone")
		}
		if !validated {
			if err := validateAssignmentPolicyReleasedTombstoneTransfer(
				ctx, store, receiptOwnerPack, receiptPolicyKey,
				permissionSetID, receiptGeneration,
			); err != nil {
				return 0, err
			}
			validatedReleaseProofs[proofKey] = receiptGeneration
		}
	}
	deleted, err := store.DeleteRows(ctx, cp.ComponentManagedDelete{
		Table: schemacatalog.AssignmentPolicyTombstoneReceiptTable,
		Match: cp.ComponentRow{"Id": receiptID}, MaximumAffectedRows: 1,
	})
	if err := assignmentPolicyRequireExactWrite(
		"delete", schemacatalog.AssignmentPolicyTombstoneReceiptTable, deleted, err,
	); err != nil {
		return 0, err
	}
	return 1, nil
}

func assignmentPolicyReleaseProofKey(ownerPack, policyKey, permissionSetID string) string {
	return string(assignmentPolicySourceIdentityBytes(ownerPack, policyKey)) + "\x00" + permissionSetID
}

func validateAssignmentPolicyReleasedTombstoneTransfer(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	ownerPack, policyKey, permissionSetID string,
	tombstonedGeneration uint64,
) error {
	if err := validateAssignmentPolicyReleasedTombstoneProof(
		ctx, store, ownerPack, policyKey, permissionSetID, tombstonedGeneration,
	); err != nil {
		return err
	}
	activeAssignments, err := store.ReadRows(ctx, assignmentPolicyQuery(
		"PermissionSetAssignment",
		[]string{"Id"},
		cp.ComponentRow{"PermissionSetId": permissionSetID, "IsDeleted": false},
		1,
		nil,
	))
	if err != nil {
		return err
	}
	if len(activeAssignments) != 0 {
		return fmt.Errorf("assignment policy tombstone target is not exact-empty")
	}
	return nil
}

func validateAssignmentPolicyReleasedTombstoneProof(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	ownerPack, policyKey, permissionSetID string,
	tombstonedGeneration uint64,
) error {
	rows, err := store.LockRows(ctx, assignmentPolicyQuery(
		schemacatalog.AssignmentPolicyReleaseReceiptTable,
		[]string{
			"Id", "OwnerPack", "PolicyKey", "TargetDomain", "PhysicalKey",
			"PermissionSetId", "ReleasedGeneration", "ExactEmptyDigest",
		},
		cp.ComponentRow{
			"OwnerPack": ownerPack, "PolicyKey": policyKey, "PermissionSetId": permissionSetID,
		},
		1,
		[]string{"OwnerPack", "PolicyKey", "PermissionSetId"},
	))
	if err != nil {
		return err
	}
	if len(rows) != 1 {
		return fmt.Errorf("assignment policy released tombstone has no exact release receipt")
	}
	actualOwnerPack, err := assignmentPolicyRequiredString(rows[0], "OwnerPack")
	if err != nil {
		return err
	}
	actualPolicyKey, err := assignmentPolicyRequiredString(rows[0], "PolicyKey")
	if err != nil {
		return err
	}
	if !assignmentPolicySourceIdentityEqual(actualOwnerPack, actualPolicyKey, ownerPack, policyKey) {
		return fmt.Errorf("assignment policy release receipt belongs to another source")
	}
	actualPermissionSetID, err := assignmentPolicyRequiredString(rows[0], "PermissionSetId")
	if err != nil {
		return err
	}
	if actualPermissionSetID != permissionSetID {
		return fmt.Errorf("assignment policy release receipt target identity differs from tombstone")
	}
	releasedGeneration, err := assignmentPolicyRequiredUint64(rows[0], "ReleasedGeneration")
	if err != nil {
		return err
	}
	if releasedGeneration == 0 {
		return fmt.Errorf("assignment policy release receipt generation is zero")
	}
	if releasedGeneration != tombstonedGeneration {
		return fmt.Errorf("assignment policy release receipt generation differs from tombstone")
	}
	permissionSetRows, err := store.ReadRows(ctx, assignmentPolicyQuery(
		"PermissionSet",
		[]string{"Id", "Name"},
		cp.ComponentRow{"Id": permissionSetID},
		1,
		[]string{"Id"},
	))
	if err != nil {
		return err
	}
	if len(permissionSetRows) != 1 {
		return fmt.Errorf("assignment policy target %s is missing", permissionSetID)
	}
	permissionSetName, err := assignmentPolicyRequiredString(permissionSetRows[0], "Name")
	if err != nil {
		return err
	}
	target, err := PermissionSetDefinitionManagedTarget(permissionSetName)
	if err != nil {
		return err
	}
	targetDomain, err := assignmentPolicyRequiredString(rows[0], "TargetDomain")
	if err != nil {
		return err
	}
	physicalKey := append([]byte(nil), assignmentPolicyBytes(rows[0]["PhysicalKey"])...)
	if targetDomain != target.TargetDomain || !bytes.Equal(physicalKey, target.PhysicalKeyBytes) {
		return fmt.Errorf("assignment policy release receipt target identity differs from live target")
	}
	if !bytes.Equal(
		assignmentPolicyBytes(rows[0]["ExactEmptyDigest"]),
		assignmentPolicyExactEmptyDigest(target, permissionSetID, releasedGeneration),
	) {
		return fmt.Errorf("assignment policy release receipt exact-empty digest is invalid")
	}
	stateRows, err := store.ReadRows(ctx, assignmentPolicyQuery(
		schemacatalog.AssignmentPolicyTargetStateTable,
		[]string{
			"Id", "TargetDomain", "PhysicalKey", "PermissionSetId", "Suspended",
			"WorkGeneration", "AppliedGeneration", "ResolutionState",
		},
		cp.ComponentRow{"PermissionSetId": permissionSetID},
		1,
		nil,
	))
	if err != nil {
		return err
	}
	if len(stateRows) != 1 {
		return fmt.Errorf("assignment policy released tombstone has no durable target state")
	}
	if err := assignmentPolicyValidateTargetStateIdentity(
		stateRows[0], permissionSetID, permissionSetName,
	); err != nil {
		return err
	}
	state, err := assignmentPolicyStateFromRow(stateRows[0])
	if err != nil {
		return err
	}
	if state.ResolutionState != assignmentPolicyResolutionRemoved ||
		state.WorkGeneration != state.AppliedGeneration || state.WorkGeneration < tombstonedGeneration {
		return fmt.Errorf("assignment policy tombstone target state is not converged removed")
	}
	return nil
}

func assignmentPolicyRequireExactWrite(operation, table string, affected int64, err error) error {
	if err != nil {
		return fmt.Errorf("assignment policy %s %s row: %w", operation, table, err)
	}
	if affected != 1 {
		return fmt.Errorf("assignment policy %s %s row affected %d, want 1", operation, table, affected)
	}
	return nil
}

func assignmentPolicyValidateTombstoneReceipt(
	row cp.ComponentRow,
	assignmentID, permissionSetID, assigneeID, ownerPack, policyKey string,
) (string, error) {
	receiptID, err := assignmentPolicyRequiredString(row, "Id")
	if err != nil {
		return "", err
	}
	actualAssignmentID, err := assignmentPolicyRequiredString(row, "AssignmentId")
	if err != nil {
		return "", err
	}
	actualPermissionSetID, err := assignmentPolicyRequiredString(row, "PermissionSetId")
	if err != nil {
		return "", err
	}
	actualAssigneeID, err := assignmentPolicyRequiredString(row, "AssigneeId")
	if err != nil {
		return "", err
	}
	actualOwnerPack, err := assignmentPolicyRequiredString(row, "OwnerPack")
	if err != nil {
		return "", err
	}
	actualPolicyKey, err := assignmentPolicyRequiredString(row, "PolicyKey")
	if err != nil {
		return "", err
	}
	if actualAssignmentID != assignmentID || actualPermissionSetID != permissionSetID ||
		actualAssigneeID != assigneeID || !assignmentPolicySourceIdentityEqual(
		actualOwnerPack, actualPolicyKey, ownerPack, policyKey,
	) {
		return "", fmt.Errorf(
			"assignment policy tombstone receipt %s does not bind exact assignment %s (%s, %s) to policy %s/%s",
			receiptID, assignmentID, permissionSetID, assigneeID, ownerPack, policyKey,
		)
	}
	return receiptID, nil
}

func readAssignmentPolicyRows(ctx context.Context, store cp.ComponentManagedReadStore, query cp.ComponentManagedRowQuery) ([]cp.ComponentRow, error) {
	if len(query.OrderBy) == 0 {
		return store.ReadRows(ctx, query)
	}
	paged, ok := store.(cp.ComponentManagedKeysetTransactionReadStore)
	if !ok {
		return nil, fmt.Errorf("assignment policy requires exact keyset read exhaustion")
	}
	var rows []cp.ComponentRow
	page := cp.ComponentManagedKeysetRowQuery{Query: query}
	for {
		result, err := paged.ReadRowsKeysetPage(ctx, page)
		if err != nil {
			return nil, err
		}
		rows = append(rows, result.Rows...)
		if result.Exhausted {
			return rows, nil
		}
		if len(result.Rows) == 0 {
			return nil, fmt.Errorf("assignment policy keyset read did not advance")
		}
		last := result.Rows[len(result.Rows)-1]
		page.ContinueAfter = make(cp.ComponentRow, len(query.OrderBy))
		for _, order := range query.OrderBy {
			value, ok := last[order.Column]
			if !ok || value == nil {
				return nil, fmt.Errorf("assignment policy keyset page is missing %s", order.Column)
			}
			page.ContinueAfter[order.Column] = value
		}
	}
}

func readAssignmentPolicyRowsUpTo(
	ctx context.Context,
	store cp.ComponentManagedReadStore,
	query cp.ComponentManagedRowQuery,
	maximumRows int,
) ([]cp.ComponentRow, bool, error) {
	paged, ok := store.(cp.ComponentManagedKeysetTransactionReadStore)
	if !ok {
		return nil, false, fmt.Errorf("assignment policy requires bounded keyset reads")
	}
	return collectAssignmentPolicyRowsUpTo(
		ctx, query, maximumRows, "read", paged.ReadRowsKeysetPage,
	)
}

type assignmentPolicyKeysetPageReader func(
	context.Context,
	cp.ComponentManagedKeysetRowQuery,
) (cp.ComponentManagedKeysetRowPage, error)

func collectAssignmentPolicyRowsUpTo(
	ctx context.Context,
	query cp.ComponentManagedRowQuery,
	maximumRows int,
	operation string,
	readPage assignmentPolicyKeysetPageReader,
) ([]cp.ComponentRow, bool, error) {
	if maximumRows < 0 {
		return nil, false, fmt.Errorf(
			"assignment policy bounded %s maximum is negative", operation,
		)
	}
	if len(query.OrderBy) == 0 {
		return nil, false, fmt.Errorf(
			"assignment policy bounded %s requires keyset order", operation,
		)
	}
	if query.MaximumRows <= 0 {
		return nil, false, fmt.Errorf(
			"assignment policy bounded %s page size is invalid", operation,
		)
	}
	var rows []cp.ComponentRow
	page := cp.ComponentManagedKeysetRowQuery{Query: query}
	for {
		page.Query.MaximumRows = min(query.MaximumRows, maximumRows+1-len(rows))
		result, err := readPage(ctx, page)
		if err != nil {
			return nil, false, err
		}
		if len(result.Rows) > page.Query.MaximumRows {
			return nil, false, fmt.Errorf(
				"assignment policy bounded %s returned %d rows; maximum is %d",
				operation, len(result.Rows), page.Query.MaximumRows,
			)
		}
		rows = append(rows, result.Rows...)
		if len(rows) > maximumRows {
			return rows, true, nil
		}
		if result.Exhausted {
			return rows, false, nil
		}
		if len(result.Rows) == 0 {
			return nil, false, fmt.Errorf(
				"assignment policy bounded %s did not advance", operation,
			)
		}
		last := result.Rows[len(result.Rows)-1]
		page.ContinueAfter = make(cp.ComponentRow, len(query.OrderBy))
		for _, order := range query.OrderBy {
			value, ok := last[order.Column]
			if !ok || value == nil {
				return nil, false, fmt.Errorf(
					"assignment policy bounded %s page is missing %s",
					operation, order.Column,
				)
			}
			page.ContinueAfter[order.Column] = value
		}
	}
}

func lockAssignmentPolicyRows(ctx context.Context, store cp.ComponentManagedMutationStore, query cp.ComponentManagedRowQuery) ([]cp.ComponentRow, error) {
	if len(query.OrderBy) == 0 {
		return store.LockRows(ctx, query)
	}
	paged, ok := store.(cp.ComponentManagedKeysetTransactionMutationStore)
	if !ok {
		return nil, fmt.Errorf("assignment policy requires exact keyset lock exhaustion")
	}
	var rows []cp.ComponentRow
	page := cp.ComponentManagedKeysetRowQuery{Query: query}
	for {
		result, err := paged.LockRowsKeysetPage(ctx, page)
		if err != nil {
			return nil, err
		}
		rows = append(rows, result.Rows...)
		if result.Exhausted {
			return rows, nil
		}
		if len(result.Rows) == 0 {
			return nil, fmt.Errorf("assignment policy locked keyset read did not advance")
		}
		last := result.Rows[len(result.Rows)-1]
		page.ContinueAfter = make(cp.ComponentRow, len(query.OrderBy))
		for _, order := range query.OrderBy {
			value, ok := last[order.Column]
			if !ok || value == nil {
				return nil, fmt.Errorf("assignment policy locked keyset page is missing %s", order.Column)
			}
			page.ContinueAfter[order.Column] = value
		}
	}
}

func lockAssignmentPolicyRowsUpTo(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	query cp.ComponentManagedRowQuery,
	maximumRows int,
) ([]cp.ComponentRow, bool, error) {
	paged, ok := store.(cp.ComponentManagedKeysetTransactionMutationStore)
	if !ok {
		return nil, false, fmt.Errorf("assignment policy requires bounded keyset locks")
	}
	return collectAssignmentPolicyRowsUpTo(
		ctx, query, maximumRows, "lock", paged.LockRowsKeysetPage,
	)
}

func assignmentPolicyRootQuery(equal cp.ComponentRow) cp.ComponentManagedRowQuery {
	return assignmentPolicyQuery(
		"PermissionSet",
		[]string{"Id", "Name", "OwnerPack", "IsActive", "IsDeleted", "IsCustom", "ProfileId", "IsOwnedByProfile"},
		equal, 1, []string{"Id"},
	)
}

func assignmentPolicyDefinitionRootFromRow(row cp.ComponentRow) (string, string, string, error) {
	id, err := assignmentPolicyRequiredString(row, "Id")
	if err != nil {
		return "", "", "", err
	}
	name, err := assignmentPolicyRequiredString(row, "Name")
	if err != nil {
		return "", "", "", err
	}
	if _, err := PermissionSetDefinitionManagedTarget(name); err != nil {
		return "", "", "", err
	}
	ownerPack, err := assignmentPolicyRequiredString(row, "OwnerPack")
	if err != nil {
		return "", "", "", err
	}
	active, err := assignmentPolicyRequiredBool(row, "IsActive")
	if err != nil {
		return "", "", "", err
	}
	deleted, err := assignmentPolicyRequiredBool(row, "IsDeleted")
	if err != nil {
		return "", "", "", err
	}
	custom, err := assignmentPolicyRequiredBool(row, "IsCustom")
	if err != nil {
		return "", "", "", err
	}
	ownedByProfile, err := assignmentPolicyRequiredBool(row, "IsOwnedByProfile")
	if err != nil {
		return "", "", "", err
	}
	profileID, profileIDPresent := row["ProfileId"]
	if !profileIDPresent {
		return "", "", "", fmt.Errorf("row is missing ProfileId")
	}
	if !active || deleted || !custom || ownedByProfile || profileID != nil {
		return "", "", "", fmt.Errorf("is not an active managed Definition")
	}
	return id, name, ownerPack, nil
}

// lockAssignmentPolicyRoots acquires the physical PermissionSet parent rows in
// the same exact-ID order as ordinary record writes. This is the cross-seam
// serialization point between policy lifecycle mutations and generic
// PermissionSetAssignment mutations.

func lockAssignmentPolicyRoots(
	ctx context.Context,
	store cp.ComponentManagedMutationStore,
	roots []assignmentPolicyRootLock,
) error {
	byID := make(map[string]assignmentPolicyRootLock, len(roots))
	for _, root := range roots {
		if root.PermissionSetID == "" || root.PermissionSet == "" || root.OwnerPack == "" {
			return fmt.Errorf("assignment policy root lock is incomplete")
		}
		if previous, exists := byID[root.PermissionSetID]; exists &&
			(!assignmentPolicyDatabaseIdentityEqual(previous.PermissionSet, root.PermissionSet) || previous.OwnerPack != root.OwnerPack) {
			return fmt.Errorf("assignment policy root %s has conflicting names", root.PermissionSetID)
		}
		byID[root.PermissionSetID] = root
	}
	ids := make([]string, 0, len(byID))
	for id := range byID {
		ids = append(ids, id)
	}
	sort.Strings(ids)
	for _, id := range ids {
		rows, err := store.LockRows(ctx, assignmentPolicyRootQuery(cp.ComponentRow{"Id": id}))
		if err != nil {
			return err
		}
		if len(rows) != 1 {
			return fmt.Errorf("assignment policy target %s changed during root locking", id)
		}
		lockedID, lockedName, lockedOwnerPack, err := assignmentPolicyDefinitionRootFromRow(rows[0])
		if err != nil {
			return fmt.Errorf("assignment policy target %s is no longer eligible after root locking: %w", id, err)
		}
		expected := byID[id]
		if lockedID != id || !assignmentPolicyDatabaseIdentityEqual(lockedName, expected.PermissionSet) ||
			lockedOwnerPack != expected.OwnerPack {
			return fmt.Errorf("assignment policy target %s changed identity during root locking", id)
		}
	}
	return nil
}

func assignmentPolicyValidateTargetStateIdentity(row cp.ComponentRow, targetID, name string) error {
	stateTargetID, err := assignmentPolicyRequiredString(row, "PermissionSetId")
	if err != nil {
		return err
	}
	stateTargetDomain, err := assignmentPolicyRequiredString(row, "TargetDomain")
	if err != nil {
		return err
	}
	statePhysicalKey := assignmentPolicyBytes(row["PhysicalKey"])
	target, err := PermissionSetDefinitionManagedTarget(name)
	if err != nil {
		return err
	}
	if stateTargetID != targetID || stateTargetDomain != target.TargetDomain ||
		!bytes.Equal(statePhysicalKey, target.PhysicalKeyBytes) {
		return fmt.Errorf("assignment policy target state identity does not match Permission Set %s", name)
	}
	return nil
}

func assignmentPolicyQuery(table string, columns []string, equal cp.ComponentRow, maximum int, logical []string) cp.ComponentManagedRowQuery {
	return cp.ComponentManagedRowQuery{
		Table: table, Columns: columns, Equal: equal,
		LogicalKeyColumns: logical, MaximumRows: maximum,
	}
}

func readAssignmentPolicyTargetNameByID(ctx context.Context, store cp.ComponentManagedReadStore, id 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 target %s is missing", id)
	}
	return assignmentPolicyRequiredString(rows[0], "Name")
}

func assignmentPolicyAssignmentFromRow(row cp.ComponentRow) (assignmentPolicyAssignmentRow, error) {
	id, err := assignmentPolicyRequiredString(row, "Id")
	if err != nil {
		return assignmentPolicyAssignmentRow{}, err
	}
	assignee, err := assignmentPolicyRequiredString(row, "AssigneeId")
	if err != nil {
		return assignmentPolicyAssignmentRow{}, err
	}
	deleted, err := assignmentPolicyRequiredBool(row, "IsDeleted")
	if err != nil {
		return assignmentPolicyAssignmentRow{}, err
	}
	return assignmentPolicyAssignmentRow{
		ID: id, Assignee: assignee, IsDeleted: deleted, LastModifiedDate: row["LastModifiedDate"],
	}, nil
}

func assignmentPolicyRequiredTimestamp(row cp.ComponentRow, column string) (any, error) {
	value, ok := row[column]
	if !ok || value == nil {
		return nil, fmt.Errorf("assignment policy row is missing %s", column)
	}
	switch typed := value.(type) {
	case time.Time:
		if typed.IsZero() {
			return nil, fmt.Errorf("assignment policy row %s is zero", column)
		}
		return typed, nil
	case string:
		if typed == "" {
			return nil, fmt.Errorf("assignment policy row %s is empty", column)
		}
		return typed, nil
	case []byte:
		if len(typed) == 0 {
			return nil, fmt.Errorf("assignment policy row %s is empty", column)
		}
		return append([]byte(nil), typed...), nil
	default:
		return nil, fmt.Errorf("assignment policy row %s has type %T", column, value)
	}
}

func assignmentPolicyTimestampsEqual(left, right any) bool {
	leftTime, leftIsTime := left.(time.Time)
	rightTime, rightIsTime := right.(time.Time)
	if leftIsTime || rightIsTime {
		return leftIsTime && rightIsTime && leftTime.Equal(rightTime)
	}
	return string(assignmentPolicyBytes(left)) == string(assignmentPolicyBytes(right))
}

func assignmentPolicyStateFromRow(row cp.ComponentRow) (assignmentPolicyStateRow, error) {
	id, err := assignmentPolicyRequiredString(row, "Id")
	if err != nil {
		return assignmentPolicyStateRow{}, err
	}
	suspended, err := assignmentPolicyRequiredBool(row, "Suspended")
	if err != nil {
		return assignmentPolicyStateRow{}, err
	}
	work, err := assignmentPolicyRequiredUint64(row, "WorkGeneration")
	if err != nil {
		return assignmentPolicyStateRow{}, err
	}
	applied, err := assignmentPolicyRequiredUint64(row, "AppliedGeneration")
	if err != nil {
		return assignmentPolicyStateRow{}, err
	}
	if applied > work {
		return assignmentPolicyStateRow{}, fmt.Errorf(
			"assignment policy applied generation %d is ahead of work generation %d",
			applied,
			work,
		)
	}
	resolution, err := assignmentPolicyRequiredString(row, "ResolutionState")
	if err != nil {
		return assignmentPolicyStateRow{}, err
	}
	return assignmentPolicyStateRow{
		ID: id, Suspended: suspended, WorkGeneration: work,
		AppliedGeneration: applied, ResolutionState: resolution,
	}, nil
}

func assignmentPolicyRequiredString(row cp.ComponentRow, column string) (string, error) {
	value, ok := row[column]
	if !ok {
		return "", fmt.Errorf("assignment policy row is missing %s", column)
	}
	switch typed := value.(type) {
	case string:
		if typed == "" {
			return "", fmt.Errorf("assignment policy row %s is empty", column)
		}
		return typed, nil
	case []byte:
		if len(typed) == 0 {
			return "", fmt.Errorf("assignment policy row %s is empty", column)
		}
		return string(typed), nil
	default:
		return "", fmt.Errorf("assignment policy row %s has type %T", column, value)
	}
}

func assignmentPolicyRequiredBool(row cp.ComponentRow, column string) (bool, error) {
	value, ok := row[column]
	if !ok {
		return false, fmt.Errorf("assignment policy row is missing %s", column)
	}
	switch typed := value.(type) {
	case bool:
		return typed, nil
	case int64:
		if typed == 0 || typed == 1 {
			return typed == 1, nil
		}
	case uint64:
		if typed == 0 || typed == 1 {
			return typed == 1, nil
		}
	case []byte:
		if parsed, err := strconv.ParseBool(string(typed)); err == nil {
			return parsed, nil
		}
	case string:
		if parsed, err := strconv.ParseBool(typed); err == nil {
			return parsed, nil
		}
	}
	return false, fmt.Errorf("assignment policy row %s has invalid boolean %v (%T)", column, value, value)
}

func assignmentPolicyRequiredUint64(row cp.ComponentRow, column string) (uint64, error) {
	value, ok := row[column]
	if !ok {
		return 0, fmt.Errorf("assignment policy row is missing %s", column)
	}
	switch typed := value.(type) {
	case uint64:
		return typed, nil
	case int64:
		if typed >= 0 {
			return uint64(typed), nil
		}
	case int:
		if typed >= 0 {
			return uint64(typed), nil
		}
	case string:
		if parsed, err := strconv.ParseUint(typed, 10, 64); err == nil {
			return parsed, nil
		}
	case []byte:
		if parsed, err := strconv.ParseUint(string(typed), 10, 64); err == nil {
			return parsed, nil
		}
	}
	return 0, fmt.Errorf("assignment policy row %s has type %T", column, value)
}

func nextAssignmentPolicyGeneration(current uint64) (uint64, error) {
	if current == math.MaxUint64 {
		return 0, fmt.Errorf("assignment policy generation is exhausted")
	}
	return current + 1, nil
}

func assignmentPolicyBytes(value any) []byte {
	switch typed := value.(type) {
	case []byte:
		return typed
	case string:
		return []byte(typed)
	default:
		return nil
	}
}

func assignmentPolicyPayloadHash(value []byte) []byte {
	digest := sha256.Sum256(value)
	return append([]byte(nil), digest[:]...)
}

func assignmentPolicyDigest(value string) []byte {
	digest := sha256.Sum256([]byte(value))
	return append([]byte(nil), digest[:]...)
}