package persistence

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"strings"

	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	"github.com/nexuscrm/shared/pkg/constants"
)

var errPermissionStateDirectWriterCoordinatorRequired = errors.New(
	"persistence: permission state direct writer coordinator is required",
)

type permissionStateDirectTarget struct {
	tableName                             string
	whereSQL                              string
	args                                  []interface{}
	prelocks                              []permissionStateDirectPrelock
	permissionSetDefinitionIntentOwnerIDs []interface{}
	permissionSetDefinitionDisposition    permissionSetDefinitionDirectMutationDisposition
	permissionSetDefinitionRootGuard      *permissionStateDirectTargetReference
}

type permissionStateDirectTargetReference struct {
	tableName string
	whereSQL  string
	args      []interface{}
}

type permissionSetDefinitionDirectMutationDisposition string

const (
	permissionSetDefinitionDirectMutationRejectManaged permissionSetDefinitionDirectMutationDisposition = "reject_managed_owner"
	permissionSetDefinitionDirectMutationDetachOnly    permissionSetDefinitionDirectMutationDisposition = "detach_only"
	permissionSetDefinitionDirectMutationCascade       permissionSetDefinitionDirectMutationDisposition = "cascade_after_root_owner_check"
)

type permissionStateDirectPrelock struct {
	tableName                         string
	whereSQL                          string
	args                              []interface{}
	lockOnly                          bool
	rejectManagedPermissionSetOwner   bool
	permissionSetDefinitionOwnerProof permissionSetDefinitionOwnerProof
}

type permissionSetDefinitionOwnerProof struct {
	kind                  permissionSetDefinitionOwnerProofKind
	targetArgumentIndexes []int
}

type permissionSetDefinitionOwnerProofKind string

const (
	permissionSetDefinitionOwnerProofSameTarget        permissionSetDefinitionOwnerProofKind = "same_target"
	permissionSetDefinitionOwnerProofCurrentProjection permissionSetDefinitionOwnerProofKind = "current_projection"
	permissionSetDefinitionOwnerProofTargetArguments   permissionSetDefinitionOwnerProofKind = "target_arguments"
	permissionSetDefinitionOwnerProofIntentOwners      permissionSetDefinitionOwnerProofKind = "intent_owners"
)

const packManagedPermissionSetDefinitionLockedMessage = "Pack-owned Permission Set definitions are immutable outside the component lifecycle provider."

func permissionStateDirectTargetFor(
	tableName string,
	whereSQL string,
	args ...interface{},
) permissionStateDirectTarget {
	return permissionStateDirectTarget{
		tableName: strings.TrimSpace(tableName),
		whereSQL:  strings.TrimSpace(whereSQL),
		args:      append([]interface{}(nil), args...),
	}
}

func permissionStateDirectTargetWithManagedPermissionSetCurrentOwnerPrelock(
	target permissionStateDirectTarget,
	parentField string,
) permissionStateDirectTarget {
	target.permissionSetDefinitionDisposition = permissionSetDefinitionDirectMutationRejectManaged
	return permissionStateDirectTargetWithPermissionSetDefinitionOwnerPrelock(
		target,
		permissionSetDefinitionOwnerProofCurrentProjection,
		true,
		permissionSetDefinitionCurrentOwnerPredicate(target, parentField),
		target.args,
		nil,
	)
}

func permissionStateDirectTargetWithManagedPermissionSetArgumentPrelock(
	target permissionStateDirectTarget,
	targetArgumentIndexes ...int,
) permissionStateDirectTarget {
	target.permissionSetDefinitionDisposition = permissionSetDefinitionDirectMutationRejectManaged
	return permissionStateDirectTargetWithPermissionSetDefinitionOwnerPrelock(
		target,
		permissionSetDefinitionOwnerProofTargetArguments,
		true,
		permissionSetDefinitionOwnerIDPredicate(len(targetArgumentIndexes)),
		permissionStateDirectTargetArguments(target, targetArgumentIndexes),
		targetArgumentIndexes,
	)
}

func permissionStateDirectTargetWithManagedPermissionSetIntentPrelock(
	target permissionStateDirectTarget,
	permissionSetIDs ...interface{},
) permissionStateDirectTarget {
	target.permissionSetDefinitionDisposition = permissionSetDefinitionDirectMutationRejectManaged
	target.permissionSetDefinitionIntentOwnerIDs = append([]interface{}(nil), permissionSetIDs...)
	return permissionStateDirectTargetWithPermissionSetDefinitionOwnerPrelock(
		target,
		permissionSetDefinitionOwnerProofIntentOwners,
		true,
		permissionSetDefinitionOwnerIDPredicate(len(permissionSetIDs)),
		permissionSetIDs,
		nil,
	)
}

func permissionStateDirectTargetWithPermissionSetDefinitionCurrentOwnerDetachPrelock(
	target permissionStateDirectTarget,
	parentField string,
) permissionStateDirectTarget {
	target.permissionSetDefinitionDisposition = permissionSetDefinitionDirectMutationDetachOnly
	return permissionStateDirectTargetWithPermissionSetDefinitionOwnerPrelock(
		target,
		permissionSetDefinitionOwnerProofCurrentProjection,
		false,
		permissionSetDefinitionCurrentOwnerPredicate(target, parentField),
		target.args,
		nil,
	)
}

func permissionStateDirectTargetWithPermissionSetDefinitionArgumentDetachPrelock(
	target permissionStateDirectTarget,
	targetArgumentIndexes ...int,
) permissionStateDirectTarget {
	target.permissionSetDefinitionDisposition = permissionSetDefinitionDirectMutationDetachOnly
	return permissionStateDirectTargetWithPermissionSetDefinitionOwnerPrelock(
		target,
		permissionSetDefinitionOwnerProofTargetArguments,
		false,
		permissionSetDefinitionOwnerIDPredicate(len(targetArgumentIndexes)),
		permissionStateDirectTargetArguments(target, targetArgumentIndexes),
		targetArgumentIndexes,
	)
}

func permissionStateDirectTargetWithPermissionSetDefinitionOwnerPrelock(
	target permissionStateDirectTarget,
	proofKind permissionSetDefinitionOwnerProofKind,
	rejectManaged bool,
	whereSQL string,
	args []interface{},
	targetArgumentIndexes []int,
) permissionStateDirectTarget {
	target.prelocks = append(target.prelocks, permissionStateDirectPrelock{
		tableName:                       constants.TablePermissionSet,
		whereSQL:                        strings.TrimSpace(whereSQL),
		args:                            append([]interface{}(nil), args...),
		rejectManagedPermissionSetOwner: rejectManaged,
		permissionSetDefinitionOwnerProof: permissionSetDefinitionOwnerProof{
			kind:                  proofKind,
			targetArgumentIndexes: append([]int(nil), targetArgumentIndexes...),
		},
	})
	return target
}

// permissionStateDirectTargetWithConnectedAppPolicyParentPrelock preserves the
// global OAuth policy lock order for a direct mutation of ConnectedAppAccess:
// Connected App first, then its access rows. Connected App itself is not a
// permission-state projection surface, so this is explicitly a lock-only
// prelock rather than a second state source.
func permissionStateDirectTargetWithConnectedAppPolicyParentPrelock(
	target permissionStateDirectTarget,
	connectedAppIDField string,
) permissionStateDirectTarget {
	target.prelocks = append(target.prelocks, permissionStateDirectPrelock{
		tableName: constants.TableConnectedApp,
		whereSQL: permissionStateDirectEqual(constants.FieldID) +
			" IN (SELECT " + quotePermissionStateDirectIdentifier(connectedAppIDField) +
			" FROM " + quotePermissionStateDirectIdentifier(target.tableName) +
			" WHERE " + target.whereSQL + ")",
		args:     append([]interface{}(nil), target.args...),
		lockOnly: true,
	})
	return target
}

func permissionStateDirectLockOnlyTableAllowed(tableName string) bool {
	switch tableName {
	case constants.TableConnectedApp, constants.TableCustomPermission, constants.TableObject:
		return true
	default:
		return false
	}
}

// permissionStateDirectTargetWithCustomPermissionCatalogPrelock takes the
// catalog row before junction mutation so grant/delete writers share the
// exact-set retirement lock order: _System_CustomPermission first, then
// PermissionSet/Profile custom-permission references.
func permissionStateDirectTargetWithCustomPermissionCatalogPrelock(
	target permissionStateDirectTarget,
	customPermissionID string,
) permissionStateDirectTarget {
	target.prelocks = append([]permissionStateDirectPrelock{{
		tableName: constants.TableCustomPermission,
		whereSQL:  permissionStateDirectEqual(constants.FieldID),
		args:      []interface{}{customPermissionID},
		lockOnly:  true,
	}}, target.prelocks...)
	return target
}

func permissionStateDirectTargetWithPermissionSetDefinitionRootCascade(
	target permissionStateDirectTarget,
	root permissionStateDirectTarget,
) permissionStateDirectTarget {
	target.permissionSetDefinitionDisposition = permissionSetDefinitionDirectMutationCascade
	target.permissionSetDefinitionRootGuard = &permissionStateDirectTargetReference{
		tableName: root.tableName,
		whereSQL:  root.whereSQL,
		args:      append([]interface{}(nil), root.args...),
	}
	return target
}

func managedPermissionSetStateTarget(
	whereSQL string,
	args ...interface{},
) permissionStateDirectTarget {
	target := permissionStateDirectTargetFor(constants.TablePermissionSet, whereSQL, args...)
	target.permissionSetDefinitionDisposition = permissionSetDefinitionDirectMutationRejectManaged
	return permissionStateDirectTargetWithPermissionSetDefinitionOwnerPrelock(
		target,
		permissionSetDefinitionOwnerProofSameTarget,
		true,
		target.whereSQL,
		target.args,
		nil,
	)
}

func permissionStateDirectTargetWithManagedPermissionSetIDPrelock(
	target permissionStateDirectTarget,
) permissionStateDirectTarget {
	return permissionStateDirectTargetWithManagedPermissionSetArgumentPrelock(target, 0)
}

func permissionStateDirectTargetWithPermissionSetDefinitionDetachIDPrelock(
	target permissionStateDirectTarget,
) permissionStateDirectTarget {
	return permissionStateDirectTargetWithPermissionSetDefinitionArgumentDetachPrelock(target, 0)
}

func permissionSetDefinitionCurrentOwnerPredicate(
	target permissionStateDirectTarget,
	parentField string,
) string {
	return quotePermissionStateDirectIdentifier(constants.FieldID) +
		" IN (SELECT " +
		quotePermissionStateDirectIdentifier(parentField) +
		" FROM " +
		quotePermissionStateDirectIdentifier(target.tableName) +
		" WHERE " +
		target.whereSQL +
		")"
}

func permissionSetDefinitionOwnerIDPredicate(count int) string {
	if count == 1 {
		return permissionStateDirectEqual(constants.FieldID)
	}
	return permissionStateDirectIn(constants.FieldID, count)
}

func permissionStateDirectTargetArguments(
	target permissionStateDirectTarget,
	indexes []int,
) []interface{} {
	args := make([]interface{}, 0, len(indexes))
	for _, index := range indexes {
		if index < 0 || index >= len(target.args) {
			return nil
		}
		args = append(args, target.args[index])
	}
	return args
}

func permissionStateDirectEqual(fieldName string) string {
	return quotePermissionStateDirectIdentifier(fieldName) + " = ?"
}

func permissionStateDirectCaseInsensitiveEqual(fieldName string) string {
	return "LOWER(" + quotePermissionStateDirectIdentifier(fieldName) + ") = LOWER(?)"
}

func permissionStateDirectIn(fieldName string, count int) string {
	placeholders := make([]string, count)
	for index := range placeholders {
		placeholders[index] = "?"
	}
	return quotePermissionStateDirectIdentifier(fieldName) +
		" IN (" + strings.Join(placeholders, ", ") + ")"
}

func (r *PermissionRepository) permissionStateDirectWriterShouldRun(
	ctx context.Context,
	writerID string,
) bool {
	if r == nil {
		return false
	}
	if permissionStateDirectWriterIsActive(ctx, writerID) {
		return false
	}
	if r.permissionStateDirectCoordinator == nil {
		return true
	}
	return r.permissionStateDirectCoordinator.shouldRun(ctx, writerID)
}

func (r *PermissionRepository) runPermissionStateDirectWriterGroup(
	ctx context.Context,
	mutation func(context.Context) error,
) error {
	coordinator := r.permissionStateDirectCoordinator
	if coordinator == nil {
		return errPermissionStateDirectWriterCoordinatorRequired
	}
	if coordinator.transactions.ownsPessimisticTx(ctx) {
		return mutation(ctx)
	}
	return coordinator.transactions.WithPessimisticTransaction(ctx, mutation)
}

func (r *PermissionRepository) runPermissionStateDirectWriter(
	ctx context.Context,
	writerID string,
	targets []permissionStateDirectTarget,
	mutation func(context.Context) error,
) error {
	coordinator := r.permissionStateDirectCoordinator
	if coordinator == nil {
		return errPermissionStateDirectWriterCoordinatorRequired
	}
	if !coordinator.shouldRun(ctx, writerID) {
		return mutation(ctx)
	}
	return coordinator.runProjected(
		ctx,
		writerID,
		targets,
		r.getExecutor,
		mutation,
	)
}

func (coordinator *PermissionStateDirectWriterCoordinator) runProjected(
	ctx context.Context,
	writerID string,
	targets []permissionStateDirectTarget,
	executor func(context.Context) Executor,
	mutation func(context.Context) error,
) error {
	if err := coordinator.validateTargets(writerID, targets); err != nil {
		return err
	}
	return coordinator.run(ctx, writerID, func(txCtx context.Context) (bool, error) {
		before, err := coordinator.loadTargetProjections(txCtx, executor(txCtx), targets, true)
		if err != nil {
			return false, err
		}
		if err := mutation(txCtx); err != nil {
			return false, err
		}
		after, err := coordinator.loadTargetProjections(txCtx, executor(txCtx), targets, false)
		if err != nil {
			return false, err
		}
		return !reflect.DeepEqual(before, after), nil
	})
}

func (coordinator *PermissionStateDirectWriterCoordinator) validateTargets(
	writerID string,
	targets []permissionStateDirectTarget,
) error {
	want, registered := coordinator.writerSurfaceIDs[writerID]
	if !registered {
		return fmt.Errorf("permission state direct writer: unregistered writer %q", writerID)
	}
	got := make(map[string]struct{})
	for _, target := range targets {
		surface, ok := coordinator.surfacesByTableName[target.tableName]
		if !ok {
			return fmt.Errorf(
				"permission state direct writer: writer %s targets unmanaged table %q",
				writerID,
				target.tableName,
			)
		}
		if target.whereSQL == "" {
			return fmt.Errorf(
				"permission state direct writer: writer %s table %s requires a bounded target predicate",
				writerID,
				target.tableName,
			)
		}
		got[surface.SurfaceID] = struct{}{}
		for _, prelock := range target.prelocks {
			_, isManagedPermissionStateSurface := coordinator.surfacesByTableName[prelock.tableName]
			if !isManagedPermissionStateSurface && !permissionStateDirectLockOnlyTableAllowed(prelock.tableName) {
				return fmt.Errorf(
					"permission state direct writer: writer %s prelocks unmanaged table %q",
					writerID,
					prelock.tableName,
				)
			}
			if prelock.lockOnly && !permissionStateDirectLockOnlyTableAllowed(prelock.tableName) {
				return fmt.Errorf(
					"permission state direct writer: writer %s uses unsupported lock-only table %q",
					writerID,
					prelock.tableName,
				)
			}
			if prelock.whereSQL == "" {
				return fmt.Errorf(
					"permission state direct writer: writer %s table %s requires a bounded prelock predicate",
					writerID,
					prelock.tableName,
				)
			}
		}
	}
	if !reflect.DeepEqual(want, got) {
		return fmt.Errorf(
			"permission state direct writer: writer %s target surfaces %v do not match inventory %v",
			writerID,
			sortedPermissionStateDirectSurfaceIDs(got),
			sortedPermissionStateDirectSurfaceIDs(want),
		)
	}
	for _, target := range targets {
		relation, definitionTarget := coordinator.permissionSetDefinitionRelationsByTableName[strings.ToLower(target.tableName)]
		if !definitionTarget || relation.Role == schemacatalog.PermissionSetDefinitionRoleSeparateAssignment {
			continue
		}
		disposition := target.permissionSetDefinitionDisposition
		targetHasRejectingPermissionSetPrelock := false
		targetHasLockOnlyPermissionSetPrelock := false
		for _, prelock := range target.prelocks {
			if prelock.tableName != constants.TablePermissionSet ||
				!permissionSetDefinitionOwnerProofMatchesTarget(target, relation, prelock) {
				continue
			}
			if prelock.rejectManagedPermissionSetOwner {
				targetHasRejectingPermissionSetPrelock = true
			} else {
				targetHasLockOnlyPermissionSetPrelock = true
			}
		}
		switch relation.Role {
		case schemacatalog.PermissionSetDefinitionRoleRoot:
			if disposition != permissionSetDefinitionDirectMutationRejectManaged {
				return fmt.Errorf(
					"permission state direct writer: writer %s table %s requires explicit managed-owner rejection",
					writerID,
					target.tableName,
				)
			}
		case schemacatalog.PermissionSetDefinitionRoleModeledGrant:
			if disposition != permissionSetDefinitionDirectMutationRejectManaged &&
				disposition != permissionSetDefinitionDirectMutationCascade {
				return fmt.Errorf(
					"permission state direct writer: writer %s table %s requires explicit managed-owner rejection or checked-root cascade",
					writerID,
					target.tableName,
				)
			}
		case schemacatalog.PermissionSetDefinitionRoleNegativeEdge:
			if disposition != permissionSetDefinitionDirectMutationRejectManaged &&
				disposition != permissionSetDefinitionDirectMutationDetachOnly &&
				disposition != permissionSetDefinitionDirectMutationCascade {
				return fmt.Errorf(
					"permission state direct writer: writer %s table %s requires explicit attach, detach, or checked-root cascade disposition",
					writerID,
					target.tableName,
				)
			}
		}
		switch disposition {
		case permissionSetDefinitionDirectMutationRejectManaged:
			if !targetHasRejectingPermissionSetPrelock {
				return fmt.Errorf(
					"permission state direct writer: writer %s table %s rejects managed ownership without its own rejecting PermissionSet prelock",
					writerID,
					target.tableName,
				)
			}
		case permissionSetDefinitionDirectMutationDetachOnly:
			if !targetHasLockOnlyPermissionSetPrelock {
				return fmt.Errorf(
					"permission state direct writer: writer %s table %s detaches without its own lock-only PermissionSet prelock",
					writerID,
					target.tableName,
				)
			}
		case permissionSetDefinitionDirectMutationCascade:
			if !coordinator.permissionSetDefinitionCascadeHasCheckedRoot(target, targets) {
				return fmt.Errorf(
					"permission state direct writer: writer %s table %s cascades without its referenced checked PermissionSet root target",
					writerID,
					target.tableName,
				)
			}
		}
	}
	return nil
}

func (coordinator *PermissionStateDirectWriterCoordinator) permissionSetDefinitionCascadeHasCheckedRoot(
	cascade permissionStateDirectTarget,
	targets []permissionStateDirectTarget,
) bool {
	reference := cascade.permissionSetDefinitionRootGuard
	if reference == nil || reference.tableName != constants.TablePermissionSet {
		return false
	}
	relation, ok := coordinator.permissionSetDefinitionRelationsByTableName[strings.ToLower(cascade.tableName)]
	if !ok || strings.TrimSpace(relation.ParentField) == "" ||
		cascade.whereSQL != permissionStateDirectEqual(relation.ParentField) ||
		!reflect.DeepEqual(cascade.args, reference.args) {
		return false
	}
	for _, candidate := range targets {
		if candidate.tableName != reference.tableName ||
			candidate.whereSQL != reference.whereSQL ||
			!reflect.DeepEqual(candidate.args, reference.args) ||
			candidate.permissionSetDefinitionDisposition != permissionSetDefinitionDirectMutationRejectManaged {
			continue
		}
		rootRelation := coordinator.permissionSetDefinitionRelationsByTableName[strings.ToLower(candidate.tableName)]
		for _, prelock := range candidate.prelocks {
			if prelock.tableName == constants.TablePermissionSet &&
				prelock.rejectManagedPermissionSetOwner &&
				permissionSetDefinitionOwnerProofMatchesTarget(candidate, rootRelation, prelock) {
				return true
			}
		}
	}
	return false
}

func permissionSetDefinitionOwnerProofMatchesTarget(
	target permissionStateDirectTarget,
	relation schemacatalog.PermissionSetDefinitionRelation,
	prelock permissionStateDirectPrelock,
) bool {
	if prelock.tableName != constants.TablePermissionSet {
		return false
	}
	switch prelock.permissionSetDefinitionOwnerProof.kind {
	case permissionSetDefinitionOwnerProofSameTarget:
		return relation.Role == schemacatalog.PermissionSetDefinitionRoleRoot &&
			prelock.whereSQL == target.whereSQL &&
			reflect.DeepEqual(prelock.args, target.args)
	case permissionSetDefinitionOwnerProofCurrentProjection:
		return strings.TrimSpace(relation.ParentField) != "" &&
			prelock.whereSQL == permissionSetDefinitionCurrentOwnerPredicate(target, relation.ParentField) &&
			reflect.DeepEqual(prelock.args, target.args)
	case permissionSetDefinitionOwnerProofTargetArguments:
		indexes := prelock.permissionSetDefinitionOwnerProof.targetArgumentIndexes
		for _, index := range indexes {
			if index < 0 || index >= len(target.args) {
				return false
			}
		}
		expectedArgs := permissionStateDirectTargetArguments(target, indexes)
		return len(indexes) > 0 &&
			permissionSetDefinitionOwnerIDsAreNonEmpty(expectedArgs) &&
			prelock.whereSQL == permissionSetDefinitionOwnerIDPredicate(len(indexes)) &&
			reflect.DeepEqual(prelock.args, expectedArgs)
	case permissionSetDefinitionOwnerProofIntentOwners:
		return len(target.permissionSetDefinitionIntentOwnerIDs) > 0 &&
			permissionSetDefinitionOwnerIDsAreNonEmpty(target.permissionSetDefinitionIntentOwnerIDs) &&
			prelock.whereSQL == permissionSetDefinitionOwnerIDPredicate(len(target.permissionSetDefinitionIntentOwnerIDs)) &&
			reflect.DeepEqual(prelock.args, target.permissionSetDefinitionIntentOwnerIDs)
	default:
		return false
	}
}

func permissionSetDefinitionOwnerIDsAreNonEmpty(ownerIDs []interface{}) bool {
	for _, ownerID := range ownerIDs {
		value, ok := ownerID.(string)
		if !ok || strings.TrimSpace(value) == "" {
			return false
		}
	}
	return true
}

func sortedPermissionStateDirectSurfaceIDs(input map[string]struct{}) []string {
	result := make([]string, 0, len(input))
	for surfaceID := range input {
		result = append(result, surfaceID)
	}
	sort.Strings(result)
	return result
}

func (coordinator *PermissionStateDirectWriterCoordinator) loadTargetProjections(
	ctx context.Context,
	exec Executor,
	targets []permissionStateDirectTarget,
	lock bool,
) ([][][]interface{}, error) {
	projections := make([][][]interface{}, len(targets))
	if lock {
		var permissionSetPrelocks []permissionStateDirectPrelock
		var otherPrelocks []permissionStateDirectPrelock
		for _, target := range targets {
			for _, prelock := range target.prelocks {
				if prelock.tableName == constants.TablePermissionSet {
					permissionSetPrelocks = append(permissionSetPrelocks, prelock)
					continue
				}
				otherPrelocks = append(otherPrelocks, prelock)
			}
		}
		// PermissionSet is the aggregate root for grant writers. Lock it before
		// shared catalogs so aggregate authoring and direct writers use the same
		// PermissionSet -> catalog -> mutation row -> Permission State order.
		if err := loadPermissionStateDefinitionPrelocks(ctx, exec, permissionSetPrelocks); err != nil {
			return nil, err
		}
		for _, prelock := range otherPrelocks {
			rows, err := loadPermissionStateDirectTarget(
				ctx,
				exec,
				prelock.tableName,
				[]string{constants.FieldID},
				permissionStateDirectTarget{
					tableName: prelock.tableName,
					whereSQL:  prelock.whereSQL,
					args:      prelock.args,
				},
				true,
			)
			if err != nil {
				return nil, err
			}
			_ = rows
		}
	}
	for index, target := range targets {
		surface := coordinator.surfacesByTableName[target.tableName]
		rows, err := loadPermissionStateDirectTarget(ctx, exec, surface.TableName, surface.SemanticFields, target, lock)
		if err != nil {
			return nil, err
		}
		projections[index] = rows
	}
	return projections, nil
}

func loadPermissionStateDefinitionPrelocks(
	ctx context.Context,
	exec Executor,
	prelocks []permissionStateDirectPrelock,
) error {
	if len(prelocks) == 0 {
		return nil
	}
	fields := []string{constants.FieldID, constants.FieldPermissionSet_OwnerPack}
	if len(prelocks) == 1 {
		rows, err := loadPermissionStateDirectTarget(
			ctx,
			exec,
			constants.TablePermissionSet,
			fields,
			permissionStateDirectTarget{
				tableName: constants.TablePermissionSet,
				whereSQL:  prelocks[0].whereSQL,
				args:      prelocks[0].args,
			},
			true,
		)
		if err != nil {
			return err
		}
		if prelocks[0].rejectManagedPermissionSetOwner &&
			permissionStateDirectRowsContainManagedPermissionSet(rows) {
			return errors.New(packManagedPermissionSetDefinitionLockedMessage)
		}
		return nil
	}

	combined := permissionStateDefinitionCombinedPrelockTarget(prelocks)
	if _, err := loadPermissionStateDirectTarget(
		ctx,
		exec,
		constants.TablePermissionSet,
		fields,
		combined,
		true,
	); err != nil {
		return err
	}
	for _, prelock := range prelocks {
		if !prelock.rejectManagedPermissionSetOwner {
			continue
		}
		rows, err := loadPermissionStateDirectTarget(
			ctx,
			exec,
			constants.TablePermissionSet,
			fields,
			permissionStateDirectTarget{
				tableName: constants.TablePermissionSet,
				whereSQL:  prelock.whereSQL,
				args:      prelock.args,
			},
			false,
		)
		if err != nil {
			return err
		}
		if permissionStateDirectRowsContainManagedPermissionSet(rows) {
			return errors.New(packManagedPermissionSetDefinitionLockedMessage)
		}
	}
	return nil
}

func permissionStateDefinitionCombinedPrelockTarget(
	prelocks []permissionStateDirectPrelock,
) permissionStateDirectTarget {
	predicates := make([]string, 0, len(prelocks))
	var args []interface{}
	for _, prelock := range prelocks {
		predicates = append(predicates, "("+prelock.whereSQL+")")
		args = append(args, prelock.args...)
	}
	return permissionStateDirectTarget{
		tableName: constants.TablePermissionSet,
		whereSQL:  strings.Join(predicates, " OR "),
		args:      args,
	}
}

func permissionStateDirectRowsContainManagedPermissionSet(
	rows [][]interface{},
) bool {
	for _, row := range rows {
		if len(row) < 2 {
			continue
		}
		if row[1] != nil {
			return true
		}
	}
	return false
}

func loadPermissionStateDirectTarget(
	ctx context.Context,
	exec Executor,
	tableName string,
	fields []string,
	target permissionStateDirectTarget,
	lock bool,
) ([][]interface{}, error) {
	quotedFields := make([]string, len(fields))
	for index, field := range fields {
		quotedFields[index] = quotePermissionStateDirectIdentifier(field)
	}
	sqlText := fmt.Sprintf(
		"SELECT %s FROM %s WHERE %s ORDER BY %s",
		strings.Join(quotedFields, ", "),
		quotePermissionStateDirectIdentifier(tableName),
		target.whereSQL,
		strings.Join(quotedFields, ", "),
	)
	if lock {
		sqlText += " FOR UPDATE"
	}
	rows, err := exec.QueryContext(ctx, sqlText, target.args...)
	if err != nil {
		return nil, fmt.Errorf(
			"permission state direct writer: load %s projection: %w",
			tableName,
			err,
		)
	}
	defer rows.Close()

	result := make([][]interface{}, 0)
	for rows.Next() {
		values := make([]interface{}, len(fields))
		destinations := make([]interface{}, len(fields))
		for index := range values {
			destinations[index] = &values[index]
		}
		if err := rows.Scan(destinations...); err != nil {
			return nil, fmt.Errorf(
				"permission state direct writer: scan %s projection: %w",
				tableName,
				err,
			)
		}
		for index, value := range values {
			if bytesValue, ok := value.([]byte); ok {
				values[index] = append([]byte(nil), bytesValue...)
			}
		}
		result = append(result, values)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf(
			"permission state direct writer: iterate %s projection: %w",
			tableName,
			err,
		)
	}
	return result, nil
}

func quotePermissionStateDirectIdentifier(identifier string) string {
	return "`" + strings.ReplaceAll(identifier, "`", "``") + "`"
}