package packs

import (
	"context"
	"database/sql"
	"fmt"
	"strings"
	"time"

	"github.com/nexuscrm/kernel/pkg/utils"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/permissions"
	schemacatalog "github.com/pingcap-inc/nexuscrm-contract/pkg/schemacatalog"
	seedrows "github.com/pingcap-inc/nexuscrm-contract/pkg/seedrows"
	"github.com/rs/zerolog/log"

	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

const fieldSysPermissionOwnerPack = "OwnerPack"

// permissionSeedScope tracks the per-profile expected coverage for business
// objects. OP and FP scopes are independent: parent-controlled children are
// excluded from OP (delegation makes child OP rows advisory) but included in
// FP (field-level security still applies per child field — AC3).
type permissionSeedScope struct {
	schemas                      []*models.ObjectMetadata
	objectPermAPINames           []string // schemas that DO get OP rows (excludes parent-controlled + derived)
	fieldPermAPINames            []string // schemas that DO get FP rows (includes parent-controlled; excludes derived)
	expectedObjectRowsPerProfile int      // == len(objectPermAPINames)
	expectedFieldRowsPerProfile  int      // sum of materializable fields across fieldPermAPINames
	// parentControlled records which schema APINames are parent-controlled,
	// so the reconcile loop can pass the same boolean flag to
	// BuildPermissionsForProfileAndSchema without re-querying metadata.
	parentControlled map[string]bool
}

// metadataParentLookup is the narrow interface the seed-scope builder uses to
// classify schemas as parent-controlled. *metadata.MetadataService implements
// it; tests inject a stub.
type metadataParentLookup interface {
	GetParentControlledObject(ctx context.Context, schema *models.ObjectMetadata) string
}

func buildBusinessPermissionSeedScope(ctx context.Context, schemas []*models.ObjectMetadata, md metadataParentLookup) permissionSeedScope {
	scope := permissionSeedScope{parentControlled: make(map[string]bool)}
	for _, schema := range schemas {
		if schema == nil || schema.IsSystem || constants.IsSystemTable(schema.APIName) || schema.UsesDerivedAccessPolicy() {
			continue
		}
		isPC := false
		if md != nil {
			isPC = md.GetParentControlledObject(ctx, schema) != ""
		}
		scope.schemas = append(scope.schemas, schema)
		// FP scope ALWAYS includes the schema (FLS independent).
		scope.fieldPermAPINames = append(scope.fieldPermAPINames, schema.APIName)
		for _, field := range schema.Fields {
			fieldCopy := field
			if permissions.ShouldMaterializeFieldPermissionRow(&fieldCopy) {
				scope.expectedFieldRowsPerProfile++
			}
		}
		// OP scope EXCLUDES parent-controlled children.
		if isPC {
			scope.parentControlled[schema.APIName] = true
			continue
		}
		scope.objectPermAPINames = append(scope.objectPermAPINames, schema.APIName)
		scope.expectedObjectRowsPerProfile++
	}
	return scope
}

func profilesNeedingPermissionReconcile(profileIDs []string, objectCounts, fieldCounts map[string]int, scope permissionSeedScope) []string {
	if scope.expectedObjectRowsPerProfile == 0 && scope.expectedFieldRowsPerProfile == 0 {
		return nil
	}

	var candidates []string
	for _, profileID := range profileIDs {
		// Source-limited (#1374): only the platform's built-in default profiles
		// are reconciled to full coverage. Imported / admin-created profiles are
		// import-/admin-owned — resolvePermRule yields no default rule for them,
		// so a reconcile pass would insert nothing; skipping them here avoids the
		// wasted work and keeps the candidate set consistent with the coverage
		// verify gate (controlplane.VerifyProfilePermissionCoverage).
		if !constants.IsPlatformDefaultProfile(profileID) {
			continue
		}
		if objectCounts[profileID] < scope.expectedObjectRowsPerProfile || fieldCounts[profileID] < scope.expectedFieldRowsPerProfile {
			candidates = append(candidates, profileID)
		}
	}
	return candidates
}

// InitializePermissions seeds default object and field permissions for all profiles.
// Permissions are stored against the Profile's backing PermissionSet (ParentId = PermSetId),
// following the Salesforce model where every Profile has an implicit PermissionSet.
//
// On warm restart (permissions already seeded), this completes in ~100ms via COUNT check.
// On cold start or schema change, it uses batch multi-value INSERT for speed.
func InitializePermissions(permSvc *permissions.PermissionService, metadata *metadata.MetadataService) error {
	log.Info().Msg("🔧 Initializing permissions...")
	totalStart := time.Now()

	// Permission seeding can process hundreds of thousands of rows on staging/prod
	// snapshots (e.g. ~474k field permissions). A 5-minute shared timeout for all
	// phases is too tight and can expire during InsertMissingFieldPermissions even
	// when the process is healthy.
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Minute)
	defer cancel()

	// Fetch all profiles from database instead of hardcoded list
	allProfiles, err := permSvc.GetAllProfiles(ctx)
	if err != nil {
		return fmt.Errorf("failed to fetch all profiles: %w", err)
	}

	profiles := make([]string, 0, len(allProfiles))
	profileNames := make(map[string]string, len(allProfiles))
	for _, p := range allProfiles {
		profiles = append(profiles, p.ID)
		profileNames[p.ID] = p.Name
	}

	// Resolve Profile → PermSetId in bulk (1-3 queries instead of 3N)
	ensureStart := time.Now()
	profileInfos := make([]struct{ ID, Name string }, 0, len(profiles))
	for _, pid := range profiles {
		profileInfos = append(profileInfos, struct{ ID, Name string }{pid, profileNames[pid]})
	}
	permSetIDs, err := permSvc.BulkEnsureProfilePermissionSets(ctx, profileInfos)
	if err != nil {
		return fmt.Errorf("BulkEnsureProfilePermissionSets: %w", err)
	}
	log.Info().Dur("elapsed", time.Since(ensureStart)).Int("profiles", len(profiles)).
		Msg("   ⏱  BulkEnsureProfilePermissionSets")

	var allSchemas []*models.ObjectMetadata
	var businessScope permissionSeedScope
	if metadata != nil {
		allSchemas = metadata.GetSchemas(ctx)
		businessScope = buildBusinessPermissionSeedScope(ctx, allSchemas, metadata)
	}

	// Bulk check which profiles need seeding (1 query instead of N)
	countStart := time.Now()
	counts, err := permSvc.BulkCountObjectPermissionsByProfiles(ctx, profiles)
	if err != nil {
		return fmt.Errorf("BulkCountObjectPermissionsByProfiles: %w", err)
	}
	unseededProfiles := make([]string, 0)
	for _, pid := range profiles {
		if counts[pid] == 0 {
			unseededProfiles = append(unseededProfiles, pid)
		}
	}
	log.Info().Dur("elapsed", time.Since(countStart)).Int("profiles", len(profiles)).
		Msg("   ⏱  BulkCountObjectPermissionsByProfiles")
	if len(unseededProfiles) == 0 {
		log.Info().Int("profiles", len(profiles)).
			Msg("   ↩ No completely unseeded profiles detected")
	} else {
		if metadata == nil {
			return fmt.Errorf("metadata service is required to seed permissions for unseeded profiles")
		}
		log.Info().Int("unseeded", len(unseededProfiles)).Int("total", len(profiles)).Msg("   🔧 Seeding permissions for new profiles")
		profiles = unseededProfiles

		// Use shared seeding helpers (same logic used by runtime permission provisioning)
		var objPerms []models.ObjectPermissions
		var fieldPerms []models.FieldPermissions

		for _, profileID := range profiles {
			psID := permSetIDs[profileID]
			for _, schema := range allSchemas {
				isPC := false
				if metadata != nil && schema != nil {
					isPC = metadata.GetParentControlledObject(ctx, schema) != ""
				}
				op, fp := permissions.BuildPermissionsForProfileAndSchema(profileID, psID, schema, isPC)
				objPerms = append(objPerms, op...)
				fieldPerms = append(fieldPerms, fp...)
			}
		}

		log.Info().Int("objectPerms", len(objPerms)).Int("fieldPerms", len(fieldPerms)).Msg("Seeding missing permissions (additive-only)")

		upsertStart := time.Now()
		if err := permSvc.InsertMissingObjectPermissions(ctx, objPerms); err != nil {
			return err
		}
		log.Info().Dur("elapsed", time.Since(upsertStart)).Int("count", len(objPerms)).
			Msg("   ⏱  InsertMissingObjectPermissions")

		fieldUpsertStart := time.Now()
		if err := permSvc.InsertMissingFieldPermissions(ctx, fieldPerms); err != nil {
			return err
		}
		log.Info().Dur("elapsed", time.Since(fieldUpsertStart)).Int("count", len(fieldPerms)).
			Msg("   ⏱  InsertMissingFieldPermissions")

		log.Info().Int("objects", len(allSchemas)).Int("profiles", len(profiles)).
			Int("objectPerms", len(objPerms)).Int("fieldPerms", len(fieldPerms)).
			Dur("total", time.Since(totalStart)).
			Msg("✅ Permissions initialized")
	}

	if err := reconcileBusinessPermissionCoverage(ctx, permSvc, profileInfos, permSetIDs, businessScope); err != nil {
		log.Warn().Err(err).Msg("Failed to reconcile partial business-object permission coverage")
	}

	return nil
}

// reconcileBusinessPermissionCoverage detects profiles whose business-object
// ObjectPermissions / FieldPermissions are only partially seeded and backfills
// the missing rows using the same shared seeding rules as profile/object creation.
func reconcileBusinessPermissionCoverage(
	ctx context.Context,
	permSvc *permissions.PermissionService,
	profileInfos []struct{ ID, Name string },
	permSetIDs map[string]string,
	scope permissionSeedScope,
) error {
	if len(profileInfos) == 0 || len(scope.schemas) == 0 {
		return nil
	}
	reconcileStart := time.Now()

	profileIDs := make([]string, 0, len(profileInfos))
	for _, profile := range profileInfos {
		if constants.IsPlatformDefaultProfile(profile.ID) {
			profileIDs = append(profileIDs, profile.ID)
		}
	}
	if len(profileIDs) == 0 {
		return nil
	}

	objectCountStart := time.Now()
	objCounts, err := permSvc.BulkCountObjectPermissionsByProfilesAndTypes(ctx, profileIDs, scope.objectPermAPINames)
	if err != nil {
		return fmt.Errorf("count business object permissions: %w", err)
	}
	log.Info().Dur("elapsed", time.Since(objectCountStart)).Int("profiles", len(profileIDs)).
		Int("object_types", len(scope.objectPermAPINames)).
		Msg("   ⏱  BulkCountObjectPermissionsByProfilesAndTypes")

	fieldCountStart := time.Now()
	fieldCounts, err := permSvc.BulkCountFieldPermissionsByProfilesAndTypes(ctx, profileIDs, scope.fieldPermAPINames)
	if err != nil {
		return fmt.Errorf("count business field permissions: %w", err)
	}
	log.Info().Dur("elapsed", time.Since(fieldCountStart)).Int("profiles", len(profileIDs)).
		Int("object_types", len(scope.fieldPermAPINames)).
		Msg("   ⏱  BulkCountFieldPermissionsByProfilesAndTypes")

	candidates := profilesNeedingPermissionReconcile(profileIDs, objCounts, fieldCounts, scope)
	if len(candidates) == 0 {
		log.Info().Dur("elapsed", time.Since(reconcileStart)).Int("profiles", len(profileIDs)).
			Int("filtered_profiles", len(profileInfos)-len(profileIDs)).
			Msg("   ↩ Business permission coverage already complete")
		return nil
	}

	log.Info().
		Int("profiles", len(candidates)).
		Int("expected_object_rows_per_profile", scope.expectedObjectRowsPerProfile).
		Int("expected_field_rows_per_profile", scope.expectedFieldRowsPerProfile).
		Msg("🔧 Reconciling partial business-object permission coverage")

	var objPerms []models.ObjectPermissions
	var fieldPerms []models.FieldPermissions
	for _, profileID := range candidates {
		psID := permSetIDs[profileID]
		for _, schema := range scope.schemas {
			isPC := schema != nil && scope.parentControlled[schema.APIName]
			op, fp := permissions.BuildPermissionsForProfileAndSchema(profileID, psID, schema, isPC)
			objPerms = append(objPerms, op...)
			fieldPerms = append(fieldPerms, fp...)
		}
	}

	// Additive-only backfill: only rows genuinely missing for a partially
	// covered profile are inserted. An imported profile's explicit values
	// (e.g. Read-only) are preserved, never overwritten with defaults.
	if len(objPerms) > 0 {
		if err := permSvc.InsertMissingObjectPermissions(ctx, objPerms); err != nil {
			return fmt.Errorf("backfill business object permissions: %w", err)
		}
	}
	if len(fieldPerms) > 0 {
		if err := permSvc.InsertMissingFieldPermissions(ctx, fieldPerms); err != nil {
			return fmt.Errorf("backfill business field permissions: %w", err)
		}
	}

	log.Info().
		Int("profiles", len(candidates)).
		Int("objectPerms", len(objPerms)).
		Int("fieldPerms", len(fieldPerms)).
		Int("filtered_profiles", len(profileInfos)-len(profileIDs)).
		Dur("elapsed", time.Since(reconcileStart)).
		Msg("✅ Reconciled partial business-object permission coverage")
	return nil
}

// SeedSystemPermissions seeds platform-owned system permissions from the
// platform-policy permission catalog. Application permissions are installed by
// their owning app packs.
func SeedSystemPermissions(ctx context.Context, db *sql.DB) error {
	pack, err := schemacatalog.StandardPermissions()
	if err != nil {
		return err
	}
	return seedSystemPermissionRows(ctx, db, pack, true, (&PlatformPolicyPack{}).Name())
}

// seedSystemPermissionRows writes catalog rows to
// _System_Permission and derives initial grants to each built-in profile's
// backing PermissionSet from the per-row defaultGrantedTo field — no
// hardcoded Go grant mapping.
func seedSystemPermissionRows(ctx context.Context, db *sql.DB, pack []seedrows.StandardPermissionRow, deactivateMissing bool, ownerPack string) error {
	if ownerPack == "" {
		return fmt.Errorf("permission catalog owner pack is required")
	}
	if err := validatePermissionCatalogRows(pack, ownerPack); err != nil {
		return err
	}
	if err := guardPermissionCatalogOwnerPack(ctx, db, pack, ownerPack); err != nil {
		return err
	}

	log.Info().Str("owner_pack", ownerPack).Msg("🔧 Seeding system permissions...")
	start := time.Now()

	log.Info().Str("owner_pack", ownerPack).Int("count", len(pack)).Msg("Seeding system permissions from pack")

	// ── Batch 1: Upsert all catalog rows in one multi-value INSERT ──
	// Writes every catalog-owned column, including RequiresPermission and
	// OwnerPack, so row-driven readers and pack drift checks use the same
	// persisted facts.
	batch1Start := time.Now()
	if len(pack) > 0 {
		var valueParts []string
		var params []interface{}
		for _, row := range pack {
			valueParts = append(valueParts, "(?, ?, ?, ?, ?, ?, ?, 1)")
			var requires interface{}
			if row.Requires != "" {
				requires = row.Requires
			}
			params = append(params,
				utils.GenerateID(), row.APIName, row.Label,
				row.Category, row.Description, requires, ownerPack,
			)
		}
		// ON DUPLICATE KEY UPDATE intentionally sets IsActive=VALUES(IsActive)
		// which is always 1 here (the VALUES clause above hardcodes 1).
		// This makes Batch 1 a true revive: a permission that was
		// deactivated by Batch 1b on a prior seed (because it was missing
		// from the pack then) flips back to IsActive=1 the moment it
		// returns to the pack. Without this, tombstoned rows would stay
		// hidden from describe/catalog/write even after the pack reinstates
		// them — reviewer's catalog-active consistency concern.
		sqlStr := fmt.Sprintf(
			"INSERT INTO `%s` (`%s`, `%s`, `%s`, `%s`, `%s`, `%s`, `%s`, `%s`) VALUES %s "+
				"ON DUPLICATE KEY UPDATE "+
				"`%s`=VALUES(`%s`), `%s`=VALUES(`%s`), `%s`=VALUES(`%s`), `%s`=VALUES(`%s`), `%s`=VALUES(`%s`), `%s`=VALUES(`%s`)",
			constants.TablePermission,
			constants.FieldID,
			constants.FieldSysPermission_APIName, constants.FieldSysPermission_Label,
			constants.FieldSysPermission_Category, constants.FieldSysPermission_Description,
			constants.FieldSysPermission_RequiresPermission,
			fieldSysPermissionOwnerPack,
			constants.FieldSysPermission_IsActive,
			strings.Join(valueParts, ", "),
			constants.FieldSysPermission_Label, constants.FieldSysPermission_Label,
			constants.FieldSysPermission_Category, constants.FieldSysPermission_Category,
			constants.FieldSysPermission_Description, constants.FieldSysPermission_Description,
			constants.FieldSysPermission_RequiresPermission, constants.FieldSysPermission_RequiresPermission,
			fieldSysPermissionOwnerPack, fieldSysPermissionOwnerPack,
			constants.FieldSysPermission_IsActive, constants.FieldSysPermission_IsActive,
		)
		if _, err := db.ExecContext(ctx, sqlStr, params...); err != nil {
			log.Warn().Err(err).Msg("Failed to batch upsert permissions")
			return err
		}
	}

	// Truth-of-pack for the catalog: any _System_Permission row whose ApiName
	// is not in the pack is deactivated (IsActive=0). This closes the
	// "removed permissions linger forever" gap flagged in review #1831 P2:
	// without this step, a permission dropped from its owning pack catalog
	// would remain visible in describe/catalog/query forever because
	// ListSystemPermissions filters IsActive=1.
	//
	// Deactivate rather than DELETE so junction rows (and any downstream
	// references in flows/audit trails) keep their foreign-key integrity.
	// Any permission that returns to the pack on a later seed flips back
	// to IsActive=1 via Batch 1's ON DUPLICATE KEY UPDATE clause (the pack
	// asserts IsActive=1, which overrides the deactivation).
	if deactivateMissing && len(pack) > 0 {
		apiNames := make([]string, 0, len(pack))
		for _, row := range pack {
			apiNames = append(apiNames, row.APIName)
		}
		placeholders := strings.Repeat("?,", len(apiNames))
		placeholders = placeholders[:len(placeholders)-1]
		deactivateSQL := fmt.Sprintf(
			"UPDATE `%s` SET `%s`=0 WHERE `%s`=1 AND `%s` = ? AND `%s` NOT IN (%s)",
			constants.TablePermission,
			constants.FieldSysPermission_IsActive,
			constants.FieldSysPermission_IsActive,
			fieldSysPermissionOwnerPack,
			constants.FieldSysPermission_APIName,
			placeholders,
		)
		deactivateArgs := make([]interface{}, 0, len(apiNames)+1)
		deactivateArgs = append(deactivateArgs, ownerPack)
		for _, n := range apiNames {
			deactivateArgs = append(deactivateArgs, n)
		}
		if res, err := db.ExecContext(ctx, deactivateSQL, deactivateArgs...); err != nil {
			log.Warn().Err(err).Msg("Failed to deactivate stale catalog rows")
			return err
		} else if n, _ := res.RowsAffected(); n > 0 {
			log.Info().Int64("deactivated", n).Msg("   ⏱  Batch 1b: catalog truth-of-pack deactivation")
		}
	}

	log.Info().Dur("elapsed", time.Since(batch1Start)).Int("count", len(pack)).
		Msg("   ⏱  Batch 1: permission definitions upsert")

	// ── Batch 2: Default grants derived from each pack row's defaultGrantedTo ──
	// Every permission carries its initial grant list inline — SystemAdmin
	// and StandardUser are the only built-in profiles the platform ships
	// with, and their grants are pure data. Adding a new permission OR
	// changing who gets it by default is a pack edit, not a code change.
	profileIDs := []string{constants.ProfileSystemAdmin, constants.ProfileStandardUser}
	permSetMap := make(map[string]string)
	for _, pid := range profileIDs {
		var permSetID string
		selectSQL := fmt.Sprintf("SELECT `%s` FROM `%s` WHERE `%s` = ? AND `%s` = 1 LIMIT 1",
			constants.FieldID, constants.TablePermissionSet,
			constants.FieldPermissionSet_ProfileID, constants.FieldPermissionSet_IsOwnedByProfile)
		if err := db.QueryRowContext(ctx, selectSQL, pid).Scan(&permSetID); err != nil {
			log.Warn().Err(err).Str("profileID", pid).Msg("Failed to resolve backing PermissionSet for profile")
			continue
		}
		permSetMap[pid] = permSetID
	}

	// Map pack-declared grant audience names to their Profile IDs. Keeps
	// the pack JSON human-readable (refers to "SystemAdmin") without
	// coupling to internal Profile-ID constants.
	audienceToProfile := map[string]string{
		"SystemAdmin":  constants.ProfileSystemAdmin,
		"StandardUser": constants.ProfileStandardUser,
	}

	type grant struct{ permSetID, permName string }
	var grants []grant
	for _, row := range pack {
		for _, audience := range row.DefaultGrantedTo {
			profileID, ok := audienceToProfile[audience]
			if !ok {
				log.Warn().Str("audience", audience).Str("permission", row.APIName).
					Msg("Unknown defaultGrantedTo audience — skipping")
				continue
			}
			psID, ok := permSetMap[profileID]
			if !ok {
				continue
			}
			grants = append(grants, grant{psID, row.APIName})
		}
	}

	// Batch 2 is ADDITIVE-ONLY: INSERT IGNORE instead of UPSERT, with no
	// "truth-of-pack" deletion pass. The pack's `defaultGrantedTo` field
	// declares INITIAL grants — the state a fresh install lands in — not
	// enforced invariants. After first install, admin edits (revokes and
	// additions) on the built-in SystemAdmin / StandardUser PermissionSets
	// are org/admin metadata and must survive reseeds.
	//
	// Why the previous UPSERT-plus-DELETE design was wrong (review #1831
	// P0): (a) ON DUPLICATE KEY UPDATE `IsEnabled`=VALUES(IsEnabled) forced
	// every seed-declared grant back to enabled, so an admin revocation of
	// a default grant would silently revert on the next boot; (b) the
	// subsequent DELETE of rows not in the pack wiped admin-added grants
	// on built-in-profile PermissionSets whenever the pack chose not to
	// declare them. Both are data-loss regressions on a seed path that
	// should only paint over blank canvases.
	//
	// New semantics — pack changes have three effects:
	//   - pack ADDS an audience for a permission   → new (ps, perm) rows
	//     are INSERTed on the next seed for any install that doesn't
	//     already have them; admin revocations survive (row exists ⇒
	//     INSERT IGNORE no-ops)
	//   - pack REMOVES an audience for a permission → no effect on any
	//     install that already seeded that grant; admins retain whatever
	//     state the row is currently in
	//   - pack removes a permission entirely        → Batch 1b deactivates
	//     the catalog row, which hides it from describe/query/write even
	//     though junction grants referencing it may still exist (tombstone
	//     semantics)
	batch2Start := time.Now()
	if len(grants) > 0 {
		var valueParts []string
		var params []interface{}
		for _, g := range grants {
			valueParts = append(valueParts, "(?, ?, ?, 1)")
			params = append(params, utils.GenerateID(), g.permSetID, g.permName)
		}
		// INSERT IGNORE: primary/unique-key collision on an existing
		// (PermissionSetId, PermissionApiName) pair silently no-ops the
		// row, preserving whatever IsEnabled / CreatedDate the row
		// currently holds.
		sqlStr := fmt.Sprintf(
			"INSERT IGNORE INTO `%s` (`%s`, `%s`, `%s`, `%s`) VALUES %s",
			constants.TablePermissionSetPermission,
			constants.FieldID,
			constants.FieldSysPermissionSetPermission_PermissionSetID,
			constants.FieldSysPermissionSetPermission_PermissionAPIName,
			constants.FieldSysPermissionSetPermission_IsEnabled,
			strings.Join(valueParts, ", "),
		)
		res, err := db.ExecContext(ctx, sqlStr, params...)
		if err != nil {
			log.Warn().Err(err).Msg("Failed to batch grant permissions (INSERT IGNORE)")
			return err
		}
		inserted, _ := res.RowsAffected()
		log.Info().Dur("elapsed", time.Since(batch2Start)).Int("declared", len(grants)).Int64("newly_inserted", inserted).
			Msg("   ⏱  Batch 2: default grants seeded (additive-only, INSERT IGNORE)")
	}

	// Runtime seeding ends here. The junction `_System_PermissionSetPermission`
	// is the sole active writer/source of truth for system grants at runtime.
	// Every other legacy source is migrated exactly once, via an additive-only
	// data-spec outside this code path:
	//
	//   - `_System_ProfilePermission`  → data-specs/2026-04-23d-migrate-profilepermission-to-junction.md
	//   - `PermissionSet.Permissions%` → data-specs/2026-04-23{a,b,c}-*.md
	//
	// Do NOT re-introduce a runtime read from any legacy source into the
	// junction — stale legacy rows would overwrite current junction state
	// (admin grants) on every boot. If an upgrade path is needed, add a
	// one-time additive-only (INSERT IGNORE) data-spec instead.

	log.Info().Str("owner_pack", ownerPack).Int("count", len(pack)).Dur("elapsed", time.Since(start)).Msg("✅ System permissions seeded")
	return nil
}

func guardPermissionCatalogOwnerPack(ctx context.Context, db *sql.DB, pack []seedrows.StandardPermissionRow, ownerPack string) error {
	if len(pack) == 0 {
		return nil
	}

	apiNames := make([]string, 0, len(pack))
	args := make([]interface{}, 0, len(pack))
	for _, row := range pack {
		apiNames = append(apiNames, row.APIName)
		args = append(args, row.APIName)
	}
	placeholders := strings.TrimRight(strings.Repeat("?,", len(apiNames)), ",")
	rows, err := db.QueryContext(ctx, fmt.Sprintf(
		"SELECT `%s`, COALESCE(`%s`, ''), `%s` FROM `%s` WHERE `%s` IN (%s) ORDER BY `%s`",
		constants.FieldSysPermission_APIName,
		fieldSysPermissionOwnerPack,
		constants.FieldSysPermission_IsActive,
		constants.TablePermission,
		constants.FieldSysPermission_APIName,
		placeholders,
		constants.FieldSysPermission_APIName,
	), args...)
	if err != nil {
		return fmt.Errorf("query permission catalog owner claims for %s: %w", ownerPack, err)
	}
	defer rows.Close()

	for rows.Next() {
		var apiName, existingOwnerPack string
		var isActive int
		if err := rows.Scan(&apiName, &existingOwnerPack, &isActive); err != nil {
			return fmt.Errorf("scan permission catalog owner claim for %s: %w", ownerPack, err)
		}
		if existingOwnerPack != "" && existingOwnerPack != ownerPack {
			return fmt.Errorf("permission %s OwnerPack = %q, want %q (IsActive=%d)", apiName, existingOwnerPack, ownerPack, isActive)
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate permission catalog owner claims for %s: %w", ownerPack, err)
	}
	return nil
}

// ValidateSourceBackedPermissionOwnerPreflight rejects source-backed permission
// declarations whose exact ApiName is already owned by a different pack before
// install plans can apply earlier metadata phases.
func ValidateSourceBackedPermissionOwnerPreflight(ctx context.Context, db *sql.DB, pack *SourceBackedPack) error {
	if pack == nil {
		return nil
	}
	rows, err := sourceBackedPermissionRows(pack.Components())
	if err != nil {
		return err
	}
	if err := guardPermissionCatalogOwnerPack(ctx, db, rows, pack.Name()); err != nil {
		return fmt.Errorf("source-backed permission owner preflight: %w", err)
	}
	return nil
}

func sourceBackedPermissionRows(components []PackComponent) ([]seedrows.StandardPermissionRow, error) {
	if len(components) == 0 {
		return nil, nil
	}
	var out []seedrows.StandardPermissionRow
	for _, component := range components {
		if component.Kind() != ComponentKindPermission {
			continue
		}
		payload, err := component.ExpectedPayload()
		if err != nil {
			return nil, fmt.Errorf("permission %s payload: %w", strings.TrimSpace(component.BusinessKey()), err)
		}
		rows, err := seedrows.ParseStandardPermissions(payload, fmt.Sprintf("permission %s payload", strings.TrimSpace(component.BusinessKey())))
		if err != nil {
			return nil, err
		}
		out = append(out, rows...)
	}
	return out, nil
}

func validatePermissionCatalogRows(pack []seedrows.StandardPermissionRow, ownerPack string) error {
	if len(pack) == 0 {
		return fmt.Errorf("permission catalog %s is empty", ownerPack)
	}
	seen := make(map[string]struct{}, len(pack))
	for _, row := range pack {
		if row.APIName == "" {
			return fmt.Errorf("permission catalog %s contains a row without apiName", ownerPack)
		}
		if _, ok := seen[row.APIName]; ok {
			return fmt.Errorf("permission catalog %s contains duplicate apiName %q", ownerPack, row.APIName)
		}
		seen[row.APIName] = struct{}{}
	}
	return nil
}

func verifySystemPermissionRows(ctx context.Context, db *sql.DB, pack []seedrows.StandardPermissionRow, ownerPack string) error {
	if ownerPack == "" {
		return fmt.Errorf("permission catalog owner pack is required")
	}
	if err := validatePermissionCatalogRows(pack, ownerPack); err != nil {
		return err
	}

	expected := make(map[string]seedrows.StandardPermissionRow, len(pack))
	for _, row := range pack {
		expected[row.APIName] = row
	}

	var errs []string
	for _, row := range pack {
		var gotLabel, gotCategory, gotDescription, gotRequires, gotOwnerPack string
		var isSystem, isActive int
		err := db.QueryRowContext(ctx, fmt.Sprintf(
			"SELECT `%s`, `%s`, COALESCE(`%s`, ''), `%s`, `%s`, COALESCE(`%s`, ''), COALESCE(`%s`, '') "+
				"FROM `%s` WHERE `%s` = ?",
			constants.FieldSysPermission_Label,
			constants.FieldSysPermission_Category,
			constants.FieldSysPermission_Description,
			constants.FieldSysPermission_IsSystem,
			constants.FieldSysPermission_IsActive,
			constants.FieldSysPermission_RequiresPermission,
			fieldSysPermissionOwnerPack,
			constants.TablePermission,
			constants.FieldSysPermission_APIName,
		), row.APIName).Scan(&gotLabel, &gotCategory, &gotDescription, &isSystem, &isActive, &gotRequires, &gotOwnerPack)
		if err == sql.ErrNoRows {
			errs = append(errs, fmt.Sprintf("permission %s is missing", row.APIName))
			continue
		}
		if err != nil {
			return fmt.Errorf("query permission %s: %w", row.APIName, err)
		}

		if gotOwnerPack != ownerPack {
			errs = append(errs, fmt.Sprintf("permission %s OwnerPack = %q, want %q", row.APIName, gotOwnerPack, ownerPack))
		}
		if isSystem == 0 {
			errs = append(errs, fmt.Sprintf("permission %s IsSystem = 0, want 1", row.APIName))
		}
		if isActive == 0 {
			errs = append(errs, fmt.Sprintf("permission %s IsActive = 0, want 1", row.APIName))
		}
		if gotLabel != row.Label {
			errs = append(errs, fmt.Sprintf("permission %s Label = %q, want %q", row.APIName, gotLabel, row.Label))
		}
		if gotCategory != row.Category {
			errs = append(errs, fmt.Sprintf("permission %s Category = %q, want %q", row.APIName, gotCategory, row.Category))
		}
		if gotDescription != row.Description {
			errs = append(errs, fmt.Sprintf("permission %s Description = %q, want %q", row.APIName, gotDescription, row.Description))
		}
		if gotRequires != row.Requires {
			errs = append(errs, fmt.Sprintf("permission %s RequiresPermission = %q, want %q", row.APIName, gotRequires, row.Requires))
		}
	}

	rows, err := db.QueryContext(ctx, fmt.Sprintf(
		"SELECT `%s` FROM `%s` WHERE COALESCE(`%s`, '') = ? AND `%s` = 1 ORDER BY `%s`",
		constants.FieldSysPermission_APIName,
		constants.TablePermission,
		fieldSysPermissionOwnerPack,
		constants.FieldSysPermission_IsActive,
		constants.FieldSysPermission_APIName,
	), ownerPack)
	if err != nil {
		return fmt.Errorf("query permission rows owned by %s: %w", ownerPack, err)
	}
	defer rows.Close()
	for rows.Next() {
		var apiName string
		if err := rows.Scan(&apiName); err != nil {
			return fmt.Errorf("scan permission row owned by %s: %w", ownerPack, err)
		}
		if _, ok := expected[apiName]; !ok {
			errs = append(errs, fmt.Sprintf("permission %s is active and owned by %s but is not declared in the payload", apiName, ownerPack))
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate permission rows owned by %s: %w", ownerPack, err)
	}

	if len(errs) > 0 {
		return fmt.Errorf("system permission catalog %s drift:\n  - %s", ownerPack, strings.Join(errs, "\n  - "))
	}
	return nil
}