package bootstrapimpl

import (
	"database/sql"
	"fmt"
	"log"
	"strings"

	"github.com/nexuscrm/db"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/assertion"
)

// AssertionViolation represents a single design violation.
//
// Relocated verbatim to nexuscrm-contract/pkg/assertion (Wave-0 SLICE B);
// this alias keeps existing importers compiling unchanged.
type AssertionViolation = assertion.AssertionViolation

// AssertionResult contains all violations found during assertion checks.
//
// Relocated verbatim to nexuscrm-contract/pkg/assertion (Wave-0 SLICE B);
// this alias keeps existing importers compiling unchanged.
type AssertionResult = assertion.AssertionResult

// RunAssertions executes all design assertions and returns violations
// By default, violations are logged as warnings. Use strictMode=true to fail on violations.
func RunAssertions(db *database.TiDBConnection, strictMode bool) (*AssertionResult, error) {
	return runAssertions(db, strictMode, false)
}

// RunPodReadinessAssertions executes the strict startup assertion set while
// tolerating the fully absent connected-delivery catalog introduced after a
// supported prior release. Partial schema/catalog state remains a violation;
// the operator-facing RunAssertions path always requires exact visibility.
func RunPodReadinessAssertions(db *database.TiDBConnection) (*AssertionResult, error) {
	return runAssertions(db, true, true)
}

func runAssertions(db *database.TiDBConnection, strictMode, allowAbsentConnectedDeliveryCatalog bool) (*AssertionResult, error) {
	log.Println("🔍 Running startup assertions...")

	result := &AssertionResult{
		Violations: []AssertionViolation{},
		Passed:     true,
	}

	// Run all assertions
	assertSystemFieldsConsistency(db.DB(), result)
	assertNoDuplicateActions(db.DB(), result)
	assertNoDuplicateFlows(db.DB(), result)
	assertNoOrphanedSharingRules(db.DB(), result)
	assertDefaultThemeExists(db.DB(), result)

	assertActiveSuperUserCapabilityHolderExists(db.DB(), result)
	assertCriticalTablesExist(db.DB(), result)
	assertManagedSystemObjectVisibilityWithPriorStatePolicy(db.DB(), result, allowAbsentConnectedDeliveryCatalog)
	assertManagedWriteValidationCapabilitiesReconciled(db.DB(), result)
	assertManagedOwnerResolutionReconciled(db.DB(), result)
	assertObjectTableMapping(db.DB(), result)
	assertFieldColumnMapping(db.DB(), result)
	assertLookupReferenceTargetsPopulated(db.DB(), result)
	assertRelationshipIntegrity(db.DB(), result)
	assertConstraintConsistency(db.DB(), result)
	assertNamingConventions(db.DB(), result)
	assertFormulaValidity(db.DB(), result)

	// Parse-time invariants on embedded JSON definitions (no DB read).
	assertWritableSystemCoreTablesHaveKeyPrefix(GetSystemTableDefinitions(), result)
	assertFlowAuditLogReadPermissionPinned(GetSystemTableDefinitions(), result)
	assertSetupAuditTrailReadPermissionPinned(GetSystemTableDefinitions(), result)
	assertAuditLogReadPermissionPinned(GetSystemTableDefinitions(), result)

	// Live-DB invariant: verify the runtime source-of-truth matches the pinned value.
	// Must run AFTER the JSON assertion so both sources are checked independently.
	assertFlowAuditLogReadPermissionLiveDB(db.DB(), result)
	assertSetupAuditTrailReadPermissionLiveDB(db.DB(), result)
	assertAuditLogReadPermissionLiveDB(db.DB(), result)
	assertRequestCaptureReadPermissionPinned(GetSystemTableDefinitions(), result)
	assertRequestCaptureReadPermissionLiveDB(db.DB(), result)
	assertRequestCaptureWritePermissionPinned(GetSystemTableDefinitions(), result)
	assertRequestCaptureWritePermissionLiveDB(db.DB(), result)
	assertExternalDeliveryAttemptReadPermissionPinned(GetSystemTableDefinitions(), result)
	assertExternalDeliveryAttemptReadPermissionLiveDB(db.DB(), result)
	assertExternalDeliveryAttemptWritePermissionPinned(GetSystemTableDefinitions(), result)
	assertExternalDeliveryAttemptWritePermissionLiveDB(db.DB(), result)

	// Runtime-required connected apps are L1/source-backed facts. App-blind L0
	// assertions must not require a product client before its owning pack is
	// installed; installed-pack verify owns those component projections.
	assertPermissionPolicySeedingComplete(db.DB(), result)
	assertDerivedArtifactsPresent(db.DB(), result)

	// Report results
	if len(result.Violations) == 0 {
		log.Println("✅ All assertions passed")
		return result, nil
	}

	result.Passed = false
	log.Printf("⚠️  Found %d assertion violation(s):", len(result.Violations))
	for i, v := range result.Violations {
		log.Printf("   %d. [%s] %s: %s", i+1, v.Severity, v.Category, v.Description)
	}

	// Check for errors
	errorCount := 0
	for _, v := range result.Violations {
		if v.Severity == constants.SeverityError {
			errorCount++
		}
	}

	if strictMode && errorCount > 0 {
		return result, apperrors.NewInternalError(fmt.Sprintf("assertion failures in strict mode: %d critical violation(s) and %d warning(s)", errorCount, len(result.Violations)-errorCount), nil)
	}

	return result, nil
}

// assertSystemFieldsConsistency checks that all MANAGED tables have required system fields.
//
// Only tables declared in system_tables.json (or returned by GetManagedTableDefinitions())
// are checked. Everything else is exempted:
//   - Discovered tables registered by DiscoverAndRegisterPhysicalTables (Id prefix "disc_")
//   - Salesforce-imported legacy tables with UUID-style _System_Object Ids
//     (e.g. addresshistory, quotehistory, userhistory, orgemailaddresssecurity)
//   - Stale managed registrations whose ApiName has since been removed from JSON
//     (e.g. _Dev_Session)
//
// Rationale: system_tables.json is the single source of truth for tables that NexusCRM
// owns. Tables outside this file have schemas defined externally (Salesforce, data
// migration, legacy imports). Many Salesforce audit/share tables legitimately lack
// CreatedDate or LastModifiedDate (append-only semantics — UserHistory has no
// LastModifiedDate by design), and enforcing the rule on them would produce false
// positives. Bootstrap only owns what it declares.
//
// The previous "Id NOT LIKE 'disc_%'" filter missed UUID-id legacy imports and stale
// `obj_` rows; switching to a positive whitelist resolves those false positives.
func assertSystemFieldsConsistency(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking system fields consistency...")

	// Required system fields for all tables
	requiredFields := constants.GetSystemFieldNames()

	// Build the managed-tables whitelist from the bootstrap definitions.
	// Keys are lowercased to match TiDB's case-insensitive name resolution.
	managedSet := make(map[string]bool)
	for _, def := range GetManagedTableDefinitions() {
		managedSet[strings.ToLower(def.TableName)] = true
	}

	// Query all objects from metadata, then filter to managed-only in Go.
	objSQL := fmt.Sprintf(
		"SELECT `%s` FROM `%s` WHERE `%s` = 0",
		constants.FieldAPIName, constants.TableObject, constants.FieldIsDeleted,
	)

	rows, err := db.Query(objSQL)
	if err != nil {
		log.Printf("   ⚠️  Could not query objects: %v", err)
		return
	}
	defer rows.Close()

	var tables []string
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err == nil {
			// Whitelist: only check tables declared in managed schema definitions.
			if managedSet[strings.ToLower(name)] {
				tables = append(tables, name)
			}
		}
	}
	if err := rows.Err(); err != nil {
		log.Printf("   ⚠️  Error iterating objects: %v", err)
		return
	}

	// Check each managed table has all required fields in metadata
	for _, tableName := range tables {
		// Query fields for this object
		fieldSQL := fmt.Sprintf(
			"SELECT f.`%s` FROM `%s` f JOIN `%s` o ON f.`%s` = o.`%s` WHERE o.`%s` = ? AND (f.`%s` = ? OR f.`%s` IS NULL)",
			constants.FieldAPIName,
			constants.TableField,
			constants.TableObject,
			constants.FieldObjectID,
			constants.ColID,
			constants.FieldAPIName,
			constants.FieldIsDeleted,
			constants.FieldIsDeleted,
		)

		fieldRows, err := db.Query(fieldSQL, tableName, false)
		if err != nil {
			continue
		}

		existingFields := make(map[string]bool)
		for fieldRows.Next() {
			var fieldName string
			if err := fieldRows.Scan(&fieldName); err == nil {
				existingFields[fieldName] = true
			}
		}
		if err := fieldRows.Err(); err != nil {
			log.Printf("   ⚠️  Error iterating fields for %s: %v", tableName, err)
		}
		_ = fieldRows.Close()

		// Check for missing required fields
		var missing []string
		for _, required := range requiredFields {
			if !existingFields[required] {
				missing = append(missing, required)
			}
		}

		if len(missing) > 0 {
			result.Violations = append(result.Violations, AssertionViolation{
				Category:    "SystemFields",
				Severity:    constants.SeverityError,
				Object:      tableName,
				Description: fmt.Sprintf("Table '%s' missing system fields in metadata: %s", tableName, strings.Join(missing, ", ")),
			})
		}
	}
}

// assertNoDuplicateActions checks that no object has duplicate action names
func assertNoDuplicateActions(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking for duplicate actions...")

	query := fmt.Sprintf(
		"SELECT `%s`, `%s`, COUNT(*) as cnt FROM `%s` GROUP BY `%s`, `%s` HAVING cnt > 1",
		constants.FieldObjectAPIName,
		constants.FieldName,
		constants.TableAction,
		constants.FieldObjectAPIName,
		constants.FieldName,
	)

	rows, err := db.Query(query)
	if err != nil {
		log.Printf("   ⚠️  Could not query actions: %v", err)
		return
	}
	defer rows.Close()

	for rows.Next() {
		var objectName, actionName string
		var count int
		if err := rows.Scan(&objectName, &actionName, &count); err == nil {
			result.Violations = append(result.Violations, AssertionViolation{
				Category:    "DuplicateAction",
				Severity:    constants.SeverityError,
				Object:      objectName,
				Description: fmt.Sprintf("Object '%s' has %d duplicate actions named '%s'", objectName, count, actionName),
			})
		}
	}
	if err := rows.Err(); err != nil {
		log.Printf("   ⚠️  Error iterating actions: %v", err)
	}
}

// assertNoDuplicateFlows checks that no object has duplicate flow triggers
func assertNoDuplicateFlows(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking for duplicate flows...")

	query := fmt.Sprintf(`SELECT fd.%s, fd.%s, COUNT(*) as cnt
		FROM %s f
		JOIN %s fd ON f.%s = fd.%s
		WHERE f.%s = 'Active' AND fd.%s != 'schedule'
		GROUP BY fd.%s, fd.%s
		HAVING cnt > 1`,
		constants.FieldFlowDefinition_TriggerObjectOrEvent,
		constants.FieldFlowDefinition_TriggerType,
		constants.TableFlow,
		constants.TableFlowDefinition,
		constants.FieldFlow_DefinitionID,
		constants.FieldFlowDefinition_ID,
		constants.FieldFlow_Status,
		constants.FieldFlowDefinition_TriggerType,
		constants.FieldFlowDefinition_TriggerObjectOrEvent,
		constants.FieldFlowDefinition_TriggerType,
	)

	rows, err := db.Query(query)
	if err != nil {
		log.Printf("   ⚠️  Could not query flows: %v", err)
		return
	}
	defer rows.Close()

	for rows.Next() {
		var objectName, triggerType string
		var count int
		if err := rows.Scan(&objectName, &triggerType, &count); err == nil {
			result.Violations = append(result.Violations, AssertionViolation{
				Category:    "DuplicateFlow",
				Severity:    constants.SeverityWarning,
				Object:      objectName,
				Description: fmt.Sprintf("Object '%s' has %d active flows with trigger '%s'", objectName, count, triggerType),
			})
		}
	}
	if err := rows.Err(); err != nil {
		log.Printf("   ⚠️  Error iterating flows: %v", err)
	}
}

// assertNoOrphanedSharingRules checks that sharing rules point to valid groups/users
func assertNoOrphanedSharingRules(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking for orphaned sharing rules...")

	query := fmt.Sprintf("SELECT sr.%s, sr.%s, sr.%s FROM %s sr LEFT JOIN `%s` g ON sr.%s = g.%s WHERE sr.%s IS NOT NULL AND g.%s IS NULL",
		constants.ColID, constants.FieldName, "ShareWithGroupId", constants.TableSharingRule, "Group", "ShareWithGroupId", constants.ColID, "ShareWithGroupId", constants.ColID)

	rows, err := db.Query(query)
	if err != nil {
		log.Printf("   ⚠️  Could not query sharing rules: %v", err)
		return
	}
	defer rows.Close()

	for rows.Next() {
		var ruleID, ruleName, groupID string
		if err := rows.Scan(&ruleID, &ruleName, &groupID); err == nil {
			result.Violations = append(result.Violations, AssertionViolation{
				Category:    "OrphanedSharingRule",
				Severity:    constants.SeverityWarning,
				Object:      ruleName,
				Description: fmt.Sprintf("Sharing rule '%s' references non-existent group '%s'", ruleName, groupID),
			})
		}
	}
	if err := rows.Err(); err != nil {
		log.Printf("   ⚠️  Error iterating sharing rules: %v", err)
	}
}

// assertDefaultThemeExists checks if a theme exists
func assertDefaultThemeExists(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking for themes...")

	var count int
	// Plain count check
	err := db.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s", constants.TableTheme)).Scan(&count)
	if err != nil {
		log.Printf("   ⚠️  Could not query themes: %v", err)
		return
	}

	if count == 0 {
		result.Violations = append(result.Violations, AssertionViolation{
			Category:    "MissingData",
			Severity:    constants.SeverityWarning,
			Object:      constants.TableTheme,
			Description: "No themes defined. UI will not load correctly.",
		})
	}
}

// assertActiveSuperUserCapabilityHolderExists ensures at least one active
// principal carries the platform super-user capability. The assertion follows
// the same SSoT tables as runtime permission materialization: profile-owned
// PermissionSets, directly assigned PermissionSets, and Permission Set Groups.
func assertActiveSuperUserCapabilityHolderExists(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking for active super-user capability holder...")

	var count int
	q := func(name string) string {
		return "`" + strings.ReplaceAll(name, "`", "``") + "`"
	}
	query := fmt.Sprintf(`
		SELECT COUNT(DISTINCT u.%s)
		FROM %s u
		WHERE u.%s = ?
		  AND (
			EXISTS (
				SELECT 1
				FROM %s ps
				JOIN %s psp
				  ON psp.%s = ps.%s
				JOIN %s perm
				  ON perm.%s = psp.%s
				 AND perm.%s = 1
				WHERE ps.%s = u.%s
				  AND ps.%s = 1
				  AND ps.%s = 0
				  AND psp.%s = ?
				  AND psp.%s = 1
			)
			OR EXISTS (
				SELECT 1
				FROM %s psa
				JOIN %s ps
				  ON ps.%s = psa.%s
				 AND ps.%s = 0
				JOIN %s psp
				  ON psp.%s = ps.%s
				JOIN %s perm
				  ON perm.%s = psp.%s
				 AND perm.%s = 1
				WHERE psa.%s = u.%s
				  AND psa.%s = 0
				  AND psp.%s = ?
				  AND psp.%s = 1
			)
			OR EXISTS (
				SELECT 1
				FROM %s psga
				JOIN %s psgc
				  ON psgc.%s = psga.%s
				JOIN %s psp
				  ON psp.%s = psgc.%s
				JOIN %s perm
				  ON perm.%s = psp.%s
				 AND perm.%s = 1
				WHERE psga.%s = u.%s
				  AND psga.%s = 0
				  AND psp.%s = ?
				  AND psp.%s = 1
				  AND NOT EXISTS (
					  SELECT 1
					  FROM %s mps
					  JOIN %s mpp
					    ON mpp.%s = mps.%s
					  WHERE mps.%s = psga.%s
					    AND mpp.%s = psp.%s
				  )
			)
		  )`,
		q(constants.ColID),
		q(constants.TableUser),
		q(constants.FieldUser_IsActive),
		q(constants.TablePermissionSet),
		q(constants.TablePermissionSetPermission),
		q(constants.FieldSysPermissionSetPermission_PermissionSetID), q(constants.ColID),
		q(constants.TablePermission),
		q(constants.FieldSysPermission_APIName), q(constants.FieldSysPermissionSetPermission_PermissionAPIName),
		q(constants.FieldSysPermission_IsActive),
		q(constants.FieldPermissionSet_ProfileID), q(constants.FieldUser_ProfileID),
		q(constants.FieldPermissionSet_IsOwnedByProfile),
		q(constants.FieldPermissionSet_IsDeleted),
		q(constants.FieldSysPermissionSetPermission_PermissionAPIName),
		q(constants.FieldSysPermissionSetPermission_IsEnabled),
		q(constants.TablePermissionSetAssignment),
		q(constants.TablePermissionSet),
		q(constants.ColID), q(constants.FieldPermissionSetAssignment_PermissionSetID),
		q(constants.FieldPermissionSet_IsDeleted),
		q(constants.TablePermissionSetPermission),
		q(constants.FieldSysPermissionSetPermission_PermissionSetID), q(constants.ColID),
		q(constants.TablePermission),
		q(constants.FieldSysPermission_APIName), q(constants.FieldSysPermissionSetPermission_PermissionAPIName),
		q(constants.FieldSysPermission_IsActive),
		q(constants.FieldPermissionSetAssignment_AssigneeID), q(constants.ColID),
		q(constants.FieldPermissionSetAssignment_IsDeleted),
		q(constants.FieldSysPermissionSetPermission_PermissionAPIName),
		q(constants.FieldSysPermissionSetPermission_IsEnabled),
		q(constants.TablePermissionSetGroupAssignment),
		q(constants.TablePermissionSetGroupComponent),
		q(constants.FieldPermissionSetGroupComponent_GroupID), q(constants.FieldPermissionSetGroupAssignment_GroupID),
		q(constants.TablePermissionSetPermission),
		q(constants.FieldSysPermissionSetPermission_PermissionSetID), q(constants.FieldPermissionSetGroupComponent_PermissionSetID),
		q(constants.TablePermission),
		q(constants.FieldSysPermission_APIName), q(constants.FieldSysPermissionSetPermission_PermissionAPIName),
		q(constants.FieldSysPermission_IsActive),
		q(constants.FieldPermissionSetGroupAssignment_AssigneeID), q(constants.ColID),
		q(constants.FieldPermissionSetGroupAssignment_IsDeleted),
		q(constants.FieldSysPermissionSetPermission_PermissionAPIName),
		q(constants.FieldSysPermissionSetPermission_IsEnabled),
		q(constants.TableMutingPermissionSet),
		q(constants.TableMutingPermissionSetPermission),
		q(constants.FieldSysMutingPermissionSetPermission_MutingPermissionSetID), q(constants.ColID),
		q(constants.FieldMutingPermissionSet_GroupID), q(constants.FieldPermissionSetGroupAssignment_GroupID),
		q(constants.FieldSysMutingPermissionSetPermission_PermissionAPIName), q(constants.FieldSysPermissionSetPermission_PermissionAPIName),
	)
	capability := string(constants.SuperUserCapability)
	err := db.QueryRow(query, true, capability, capability, capability).Scan(&count)
	if err != nil {
		log.Printf("   ⚠️  Could not query active super-user capability holders: %v", err)
		return
	}

	if count == 0 {
		result.Violations = append(result.Violations, AssertionViolation{
			Category:    "MissingData",
			Severity:    constants.SeverityError,
			Object:      constants.TableUser,
			Description: fmt.Sprintf("No active users carry super-user capability %q. You may be locked out.", capability),
		})
	}
}

// assertCriticalTablesExist checks for presence of core system tables
func assertCriticalTablesExist(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking for critical system tables...")

	tables := []string{
		constants.TableObject,
		constants.TableField,
		constants.TableProfile,
		constants.TableUser,
		constants.TableApp,
	}

	for _, t := range tables {
		// Use Raw SQL to avoid builder backticking "1" as a column
		query := fmt.Sprintf("SELECT 1 FROM %s LIMIT 1", t)
		_, err := db.Exec(query)
		if err != nil {
			// Check if error is actually "table doesn't exist" vs "empty table" (which Select 1 handles fine usually, but Exec might return error if no rows?)
			// Actually DB.Exec on Select might return result with 0 rows affected, and nil error.
			// Only if table doesn't exist, it returns specific error.

			// Wait, if table is empty, SELECT 1 LIMIT 1 returns 0 rows. Exec returns nil error?
			// Yes.

			// However, verify specifically for Table doesn't exist error code if possible, or just assume error = missing.
			result.Violations = append(result.Violations, AssertionViolation{
				Category:    "MissingTable",
				Severity:    constants.SeverityError,
				Object:      t,
				Description: fmt.Sprintf("Critical system table '%s' is missing from the database (Error: %v).", t, err),
			})
		}
	}
}

// assertManagedSystemObjectVisibility checks that bootstrap-owned system
// objects are present and not soft-deleted in _System_Object.
//
// This catches the specific drift mode that presents as a 404 on
// `/sobjects/_System_File/describe`: the physical table still exists, but the
// metadata row was hidden by `IsDeleted = 1`.
func assertManagedSystemObjectVisibility(db *sql.DB, result *AssertionResult) {
	assertManagedSystemObjectVisibilityWithPriorStatePolicy(db, result, false)
}

func assertManagedSystemObjectVisibilityWithPriorStatePolicy(db *sql.DB, result *AssertionResult, allowAbsentConnectedDeliveryCatalog bool) {
	log.Println("   📋 Checking managed system object visibility...")

	type rowState struct {
		exists    bool
		isDeleted bool
	}

	rows, err := db.Query(
		fmt.Sprintf("SELECT `%s`, `%s` FROM `%s`", constants.FieldAPIName, constants.FieldIsDeleted, constants.TableObject),
	)
	if err != nil {
		log.Printf("   ⚠️  Could not query managed object visibility: %v", err)
		return
	}
	defer rows.Close()

	visibility := make(map[string]rowState)
	for rows.Next() {
		var apiName string
		var deleted sql.NullBool
		if err := rows.Scan(&apiName, &deleted); err != nil {
			continue
		}
		visibility[strings.ToLower(apiName)] = rowState{
			exists:    true,
			isDeleted: deleted.Valid && deleted.Bool,
		}
	}
	if err := rows.Err(); err != nil {
		log.Printf("   ⚠️  Error iterating managed object visibility rows: %v", err)
		return
	}

	deferAbsentConnectedDeliveryCatalog := false
	if allowAbsentConnectedDeliveryCatalog {
		allMetadataRowsAbsent := true
		for _, tableName := range connectedDeliveryTableNames {
			if _, ok := visibility[strings.ToLower(tableName)]; ok {
				allMetadataRowsAbsent = false
				break
			}
		}
		if allMetadataRowsAbsent {
			allPhysicalTablesAbsent, err := connectedDeliveryPhysicalTablesAbsent(db)
			if err != nil {
				log.Printf("   ⚠️  Could not classify supported prior connected-delivery state: %v", err)
			} else {
				deferAbsentConnectedDeliveryCatalog = allPhysicalTablesAbsent
			}
		}
	}

	for _, def := range GetManagedTableDefinitions() {
		state, ok := visibility[strings.ToLower(def.TableName)]
		if !ok && deferAbsentConnectedDeliveryCatalog && isConnectedDeliveryManagedTable(def.TableName) {
			continue
		}
		if !ok || state.isDeleted {
			result.Violations = append(result.Violations, AssertionViolation{
				Category:    "MissingData",
				Severity:    constants.SeverityError,
				Object:      def.TableName,
				Description: fmt.Sprintf("Managed object '%s' is missing or soft-deleted in _System_Object; run setup apply to restore visibility.", def.TableName),
			})
		}
	}
}

func connectedDeliveryPhysicalTablesAbsent(db *sql.DB) (bool, error) {
	if len(connectedDeliveryTableNames) == 0 {
		return false, nil
	}
	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(connectedDeliveryTableNames)), ",")
	args := make([]any, len(connectedDeliveryTableNames))
	for i, tableName := range connectedDeliveryTableNames {
		args[i] = tableName
	}
	var count int
	if err := db.QueryRow(
		"SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME IN ("+placeholders+")",
		args...,
	).Scan(&count); err != nil {
		return false, err
	}
	return count == 0, nil
}

// Metadata/Schema assertion functions are in assertions_metadata.go:
// - assertObjectTableMapping, assertFieldColumnMapping
// - assertRelationshipIntegrity, assertConstraintConsistency
// - assertNamingConventions, assertFormulaValidity

// assertFlowAuditLogReadPermissionPinned is a parse-time invariant on the
// embedded system_tables.json definitions (no DB read).
//
// _System_FlowAuditLog audit rows carry L3 record values in payload columns
// (InputValues, OutputValues, ErrorMessage). Exposing these rows to any user
// with ViewSetup would leak record data to setup-only admins who lack
// ViewAllData. The fix (issue #503, Option C-extended) pins the read gate
// to "ViewAllData,ModifyAllData" — the same privileged-data tier that
// record_access.go treats as equivalent for all L3 access purposes.
//
// This assertion fires on every RunAssertions call, which is reached by both
// controlplane.Verify (verify.go) and the ApplyAll path (verify_l0.go). If
// the JSON value drifts (e.g. reverted to "ViewSetup" by a merge conflict),
// the server refuses to start in strict mode before any request is served.
//
// Drift exposure window: ApplyAll is operator-driven only — invoked via
// "cmd/setup apply" (cmd/setup/main.go:58). There is no cron or per-deploy
// automatic invocation. The drift-exposure bound for the JSON-pin assertion
// is therefore "until the next operator-driven cmd/setup apply run". The
// live-DB assertion below (assertFlowAuditLogReadPermissionLiveDB) closes
// the JSON-vs-DB gap by checking the actual row at every server startup
// (strict-mode refuses startup on violation). The generic-CRUD downgrade
// vector is closed by including RequiredReadPermission in
// records.immutableObjectFields (persistence_service.go:1549). Residual
// exposure: only direct SQL writes (DBA shell) bypass both guards; next
// startup or cmd/setup apply will detect and refuse / reconcile.
const flowAuditLogRequiredReadPermission = "ViewAllData,ModifyAllData"
const setupAuditTrailRequiredReadPermission = "ViewSetupAuditTrail,ViewAllData,ModifyAllData"
const auditLogRequiredReadPermission = "ViewAllData,ModifyAllData"
const requestCaptureRequiredReadPermission = "ViewAllData,ModifyAllData"
const requestCaptureRequiredWritePermission = "BLOCKED"
const externalDeliveryAttemptRequiredReadPermission = "BLOCKED"
const externalDeliveryAttemptRequiredWritePermission = "BLOCKED"
const connectorEventRequiredReadPermission = "BLOCKED"
const connectorEventRequiredWritePermission = "BLOCKED"

func assertFlowAuditLogReadPermissionPinned(defs []schema.TableDefinition, result *AssertionResult) {
	log.Println("   📋 Checking _System_FlowAuditLog requiredReadPermission pin...")

	for _, def := range defs {
		if def.TableName != "_System_FlowAuditLog" {
			continue
		}
		if def.RequiredReadPermission != flowAuditLogRequiredReadPermission {
			result.Violations = append(result.Violations, AssertionViolation{
				Category: "AuditReadGate",
				Severity: constants.SeverityError,
				Object:   "_System_FlowAuditLog",
				Description: fmt.Sprintf(
					`_System_FlowAuditLog.requiredReadPermission must be %q (got %q); `+
						`audit-payload columns carry L3 record values and require VAD/MAD gating per #503`,
					flowAuditLogRequiredReadPermission, def.RequiredReadPermission,
				),
			})
		}
		return
	}

	// Table missing entirely — also a violation.
	result.Violations = append(result.Violations, AssertionViolation{
		Category:    "AuditReadGate",
		Severity:    constants.SeverityError,
		Object:      "_System_FlowAuditLog",
		Description: "_System_FlowAuditLog entry missing from system_tables.json; cannot verify audit read gate",
	})
}

func assertSetupAuditTrailReadPermissionPinned(defs []schema.TableDefinition, result *AssertionResult) {
	log.Println("   📋 Checking _System_SetupAuditTrail requiredReadPermission pin...")

	for _, def := range defs {
		if def.TableName != "_System_SetupAuditTrail" {
			continue
		}
		if def.RequiredReadPermission != setupAuditTrailRequiredReadPermission {
			result.Violations = append(result.Violations, AssertionViolation{
				Category: "AuditReadGate",
				Severity: constants.SeverityError,
				Object:   "_System_SetupAuditTrail",
				Description: fmt.Sprintf(
					`_System_SetupAuditTrail.requiredReadPermission must be %q (got %q); `+
						`setup-trail entries expose admin actions on private metadata and require `+
						`ViewSetupAuditTrail/VAD/MAD gating per #604`,
					setupAuditTrailRequiredReadPermission, def.RequiredReadPermission,
				),
			})
		}
		return
	}

	result.Violations = append(result.Violations, AssertionViolation{
		Category:    "AuditReadGate",
		Severity:    constants.SeverityError,
		Object:      "_System_SetupAuditTrail",
		Description: "_System_SetupAuditTrail entry missing from system_tables.json; cannot verify audit read gate",
	})
}

// assertAuditLogReadPermissionPinned is a parse-time invariant on the
// embedded system_tables.json definitions (no DB read).
//
// _System_AuditLog rows carry L3 record values in payload columns
// (OldValue, NewValue, ChangesJSON, ClientIp, SessionId, ActualUserId).
// Exposing these rows to any authenticated user via generic SOQL would leak
// field-level change history and session forensics data. The fix (issue #600,
// Option C-extended, sibling to #503/#564) pins the read gate to
// "ViewAllData,ModifyAllData" — the same privileged-data tier used for
// _System_FlowAuditLog.
func assertAuditLogReadPermissionPinned(defs []schema.TableDefinition, result *AssertionResult) {
	log.Println("   📋 Checking _System_AuditLog requiredReadPermission pin...")

	for _, def := range defs {
		if def.TableName != "_System_AuditLog" {
			continue
		}
		if def.RequiredReadPermission != auditLogRequiredReadPermission {
			result.Violations = append(result.Violations, AssertionViolation{
				Category: "AuditReadGate",
				Severity: constants.SeverityError,
				Object:   "_System_AuditLog",
				Description: fmt.Sprintf(
					`_System_AuditLog.requiredReadPermission must be %q (got %q); `+
						`audit-payload columns carry L3 record values and require VAD/MAD gating per #600`,
					auditLogRequiredReadPermission, def.RequiredReadPermission,
				),
			})
		}
		return
	}

	// Table missing entirely — also a violation.
	result.Violations = append(result.Violations, AssertionViolation{
		Category:    "AuditReadGate",
		Severity:    constants.SeverityError,
		Object:      "_System_AuditLog",
		Description: "_System_AuditLog entry missing from system_tables.json; cannot verify audit read gate",
	})
}

// assertFlowAuditLogReadPermissionLiveDB enforces the same invariant as
// assertFlowAuditLogReadPermissionPinned, but against the LIVE database row that
// the runtime permission gate actually reads (schema.RequiredReadPermission
// is populated from _System_Object via metadata_scan_schema.go). This catches
// the drift mode that JSON-only checks miss: an existing deployment whose
// _System_FlowAuditLog row still carries RequiredReadPermission='ViewSetup'
// will pass the JSON assertion (which checks system_tables.json, not the DB)
// but expose audit-payload rows to setup-only users at runtime until an
// operator runs setup apply.
//
// Edge case — row absent (fresh DB before first ApplyAll):
//
//	Skip the assertion (return nil). There is nothing to enforce yet; the
//	ApplyAll path will seed the correct value. Failing a fresh-DB startup would
//	create a bootstrapping deadlock. This matches the spirit of the JSON
//	assertion, which only fires on a present-but-wrong value (missing JSON
//	entry is also flagged, but the DB case is different: a brand-new DB has
//	no rows yet, and the server must be able to start in order to run ApplyAll).
func assertFlowAuditLogReadPermissionLiveDB(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking _System_FlowAuditLog RequiredReadPermission live DB row...")

	var liveValue sql.NullString
	err := db.QueryRow(
		fmt.Sprintf(
			"SELECT `%s` FROM `%s` WHERE `%s` = ? LIMIT 1",
			constants.FieldSysObject_RequiredReadPermission,
			constants.TableObject,
			constants.FieldAPIName,
		),
		"_System_FlowAuditLog",
	).Scan(&liveValue)

	if err == sql.ErrNoRows {
		// Fresh DB — row not yet seeded. Skip; ApplyAll will reconcile.
		log.Println("   ℹ️  _System_FlowAuditLog row not found in DB (fresh DB?); skipping live-row check")
		return
	}
	if err != nil {
		log.Printf("   ⚠️  Could not query _System_Object for _System_FlowAuditLog: %v", err)
		return
	}

	live := liveValue.String // NullString.String is "" when NULL — both are wrong.
	if live != flowAuditLogRequiredReadPermission {
		result.Violations = append(result.Violations, AssertionViolation{
			Category: "AuditReadGate",
			Severity: constants.SeverityError,
			Object:   "_System_FlowAuditLog",
			Description: fmt.Sprintf(
				`_System_Object live row for _System_FlowAuditLog has RequiredReadPermission=%q (want %q); `+
					`the DB has drifted from system_tables.json — runtime audit-payload gate is using the wrong value. `+
					`Run setup apply to reconcile, or restore the value via DB if reconciliation has been deliberately disabled.`,
				live, flowAuditLogRequiredReadPermission,
			),
		})
	}
}

func assertSetupAuditTrailReadPermissionLiveDB(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking _System_SetupAuditTrail RequiredReadPermission live DB row...")

	var liveValue sql.NullString
	err := db.QueryRow(
		fmt.Sprintf(
			"SELECT `%s` FROM `%s` WHERE `%s` = ? LIMIT 1",
			constants.FieldSysObject_RequiredReadPermission,
			constants.TableObject,
			constants.FieldAPIName,
		),
		"_System_SetupAuditTrail",
	).Scan(&liveValue)

	if err == sql.ErrNoRows {
		log.Println("   ℹ️  _System_SetupAuditTrail row not found in DB (fresh DB?); skipping live-row check")
		return
	}
	if err != nil {
		log.Printf("   ⚠️  Could not query _System_Object for _System_SetupAuditTrail: %v", err)
		return
	}

	live := liveValue.String
	if live != setupAuditTrailRequiredReadPermission {
		result.Violations = append(result.Violations, AssertionViolation{
			Category: "AuditReadGate",
			Severity: constants.SeverityError,
			Object:   "_System_SetupAuditTrail",
			Description: fmt.Sprintf(
				`_System_Object live row for _System_SetupAuditTrail has RequiredReadPermission=%q (want %q); `+
					`the DB has drifted from system_tables.json — runtime audit-setup-trail gate is using the wrong value. `+
					`Run setup apply to reconcile.`,
				live, setupAuditTrailRequiredReadPermission,
			),
		})
	}
}

// assertAuditLogReadPermissionLiveDB enforces the same invariant as
// assertAuditLogReadPermissionPinned, but against the LIVE database row that
// the runtime permission gate actually reads. This catches drift: an existing
// deployment whose _System_AuditLog row still carries RequiredReadPermission=”
// will pass the JSON assertion but expose audit rows to all authenticated users
// at runtime until an operator runs setup apply.
//
// Edge case — row absent (fresh DB before first ApplyAll): skip.
func assertAuditLogReadPermissionLiveDB(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking _System_AuditLog RequiredReadPermission live DB row...")

	var liveValue sql.NullString
	err := db.QueryRow(
		fmt.Sprintf(
			"SELECT `%s` FROM `%s` WHERE `%s` = ? LIMIT 1",
			constants.FieldSysObject_RequiredReadPermission,
			constants.TableObject,
			constants.FieldAPIName,
		),
		"_System_AuditLog",
	).Scan(&liveValue)

	if err == sql.ErrNoRows {
		log.Println("   ℹ️  _System_AuditLog row not found in DB (fresh DB?); skipping live-row check")
		return
	}
	if err != nil {
		log.Printf("   ⚠️  Could not query _System_Object for _System_AuditLog: %v", err)
		return
	}

	live := liveValue.String
	if live != auditLogRequiredReadPermission {
		result.Violations = append(result.Violations, AssertionViolation{
			Category: "AuditReadGate",
			Severity: constants.SeverityError,
			Object:   "_System_AuditLog",
			Description: fmt.Sprintf(
				`_System_Object live row for _System_AuditLog has RequiredReadPermission=%q (want %q); `+
					`the DB has drifted from system_tables.json — runtime audit-payload gate is using the wrong value. `+
					`Run setup apply to reconcile, or restore the value via DB if reconciliation has been deliberately disabled.`,
				live, auditLogRequiredReadPermission,
			),
		})
	}
}

// assertRequestCaptureReadPermissionPinned is a parse-time invariant on the
// embedded system_tables.json definitions (no DB read).
//
// _System_RequestCapture audit rows carry L3 record values AND secrets in
// payload columns (Body, ResponseBody, Headers — which include auth headers and
// API keys). Exposing these rows to any user without ViewAllData/ModifyAllData
// would leak both record data and credentials. Issue #601 pins the read gate
// to "ViewAllData,ModifyAllData" — the same privileged-data tier that
// record_access.go treats as equivalent for all L3 access purposes.
// audit-payload columns carry L3 record values + secrets (auth headers, API keys)
// and require VAD/MAD gating per #601.
func assertRequestCaptureReadPermissionPinned(defs []schema.TableDefinition, result *AssertionResult) {
	log.Println("   📋 Checking _System_RequestCapture requiredReadPermission pin...")

	for _, def := range defs {
		if def.TableName != "_System_RequestCapture" {
			continue
		}
		if def.RequiredReadPermission != requestCaptureRequiredReadPermission {
			result.Violations = append(result.Violations, AssertionViolation{
				Category: "AuditReadGate",
				Severity: constants.SeverityError,
				Object:   "_System_RequestCapture",
				Description: fmt.Sprintf(
					`_System_RequestCapture.requiredReadPermission must be %q (got %q); `+
						`audit-payload columns carry L3 record values + secrets (auth headers, API keys) and require VAD/MAD gating per #601`,
					requestCaptureRequiredReadPermission, def.RequiredReadPermission,
				),
			})
		}
		return
	}

	// Table missing entirely — also a violation.
	result.Violations = append(result.Violations, AssertionViolation{
		Category:    "AuditReadGate",
		Severity:    constants.SeverityError,
		Object:      "_System_RequestCapture",
		Description: "_System_RequestCapture entry missing from system_tables.json; cannot verify audit read gate",
	})
}

// assertRequestCaptureReadPermissionLiveDB enforces the same invariant as
// assertRequestCaptureReadPermissionPinned, but against the LIVE database row
// that the runtime permission gate actually reads. This catches the drift mode
// that JSON-only checks miss: an existing deployment whose _System_RequestCapture
// row still carries an open RequiredReadPermission will pass the JSON assertion
// but expose audit-payload rows (including secrets) to unprivileged users at
// runtime until an operator runs setup apply.
//
// Edge case — row absent (fresh DB before first ApplyAll): skip the assertion
// (return nil). There is nothing to enforce yet; ApplyAll will seed the correct
// value. Failing a fresh-DB startup would create a bootstrapping deadlock.
func assertRequestCaptureReadPermissionLiveDB(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking _System_RequestCapture RequiredReadPermission live DB row...")

	var liveValue sql.NullString
	err := db.QueryRow(
		fmt.Sprintf(
			"SELECT `%s` FROM `%s` WHERE `%s` = ? LIMIT 1",
			constants.FieldSysObject_RequiredReadPermission,
			constants.TableObject,
			constants.FieldAPIName,
		),
		"_System_RequestCapture",
	).Scan(&liveValue)

	if err == sql.ErrNoRows {
		log.Println("   ℹ️  _System_RequestCapture row not found in DB (fresh DB?); skipping live-row RRP check")
		return
	}
	if err != nil {
		log.Printf("   ⚠️  Could not query _System_Object for _System_RequestCapture: %v", err)
		return
	}

	live := liveValue.String // NullString.String is "" when NULL — both are wrong.
	if live != requestCaptureRequiredReadPermission {
		result.Violations = append(result.Violations, AssertionViolation{
			Category: "AuditReadGate",
			Severity: constants.SeverityError,
			Object:   "_System_RequestCapture",
			Description: fmt.Sprintf(
				`_System_Object live row for _System_RequestCapture has RequiredReadPermission=%q (want %q); `+
					`the DB has drifted from system_tables.json — runtime audit-payload gate (incl. secrets) is using the wrong value. `+
					`Run setup apply to reconcile, or restore the value via DB if reconciliation has been deliberately disabled.`,
				live, requestCaptureRequiredReadPermission,
			),
		})
	}
}

// assertRequestCaptureWritePermissionPinned is a parse-time invariant on the
// embedded system_tables.json definitions (no DB read).
//
// _System_RequestCapture must be "BLOCKED" for writes via generic CRUD.
// The capture middleware writes directly via SQL (bypassing the CRUD layer),
// so blocking generic writes does not break capture. This prevents any user
// or automated flow from overwriting or injecting rows via the generic API,
// which could corrupt the capture audit trail or inject misleading entries.
func assertRequestCaptureWritePermissionPinned(defs []schema.TableDefinition, result *AssertionResult) {
	log.Println("   📋 Checking _System_RequestCapture requiredWritePermission pin...")

	for _, def := range defs {
		if def.TableName != "_System_RequestCapture" {
			continue
		}
		if def.RequiredWritePermission != requestCaptureRequiredWritePermission {
			result.Violations = append(result.Violations, AssertionViolation{
				Category: "AuditWriteGate",
				Severity: constants.SeverityError,
				Object:   "_System_RequestCapture",
				Description: fmt.Sprintf(
					`_System_RequestCapture.requiredWritePermission must be %q (got %q); `+
						`requiredWritePermission must be "BLOCKED" to prevent generic CRUD writes to secrets-bearing payload`,
					requestCaptureRequiredWritePermission, def.RequiredWritePermission,
				),
			})
		}
		return
	}

	// Table missing entirely — also a violation.
	result.Violations = append(result.Violations, AssertionViolation{
		Category:    "AuditWriteGate",
		Severity:    constants.SeverityError,
		Object:      "_System_RequestCapture",
		Description: "_System_RequestCapture entry missing from system_tables.json; cannot verify audit write gate",
	})
}

// assertRequestCaptureWritePermissionLiveDB enforces the same invariant as
// assertRequestCaptureWritePermissionPinned, but against the LIVE database row.
// This catches drift where the DB row has been modified to allow writes after
// the JSON-level gate was set to BLOCKED.
//
// Edge case — row absent (fresh DB before first ApplyAll): skip the assertion
// (return nil) to avoid a bootstrapping deadlock.
func assertRequestCaptureWritePermissionLiveDB(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking _System_RequestCapture RequiredWritePermission live DB row...")

	var liveValue sql.NullString
	err := db.QueryRow(
		fmt.Sprintf(
			"SELECT `%s` FROM `%s` WHERE `%s` = ? LIMIT 1",
			constants.FieldSysObject_RequiredWritePermission,
			constants.TableObject,
			constants.FieldAPIName,
		),
		"_System_RequestCapture",
	).Scan(&liveValue)

	if err == sql.ErrNoRows {
		log.Println("   ℹ️  _System_RequestCapture row not found in DB (fresh DB?); skipping live-row WRP check")
		return
	}
	if err != nil {
		log.Printf("   ⚠️  Could not query _System_Object WRP for _System_RequestCapture: %v", err)
		return
	}

	live := liveValue.String // NullString.String is "" when NULL — both are wrong.
	if live != requestCaptureRequiredWritePermission {
		result.Violations = append(result.Violations, AssertionViolation{
			Category: "AuditWriteGate",
			Severity: constants.SeverityError,
			Object:   "_System_RequestCapture",
			Description: fmt.Sprintf(
				`_System_Object live row for _System_RequestCapture has RequiredWritePermission=%q (want %q); `+
					`the DB has drifted from system_tables.json — generic CRUD write gate for secrets-bearing payload is using the wrong value. `+
					`Run setup apply to reconcile, or restore the value via DB if reconciliation has been deliberately disabled.`,
				live, requestCaptureRequiredWritePermission,
			),
		})
	}
}

// assertExternalDeliveryAttemptReadPermissionPinned is a parse-time invariant on
// the embedded system_tables.json definitions (no DB read).
//
// _System_ExternalDeliveryAttempt rows contain frozen email bodies/payloads,
// recipient details, and target endpoints that may include L3 record data and
// delivery secrets. Generic CRUD read is BLOCKED entirely; the only read surface
// is the delivery admin API, which requires the dedicated ViewOutboundDeliveries/
// ResendOutboundDeliveries permission at the route AND per-source-record read
// access (RetrieveWithAccess) in the handler — mirroring the connector
// sync-status surface. A permission holder cannot inspect deliveries for records
// they cannot otherwise see.
//
// Delegated operators are record-scoped; system admins and ViewAllData/ModifyAllData
// holders reach evidence directly — including attempts whose source record was rolled
// back or deleted (evidence is persisted outside the business transaction for exactly
// this reason).
func assertExternalDeliveryAttemptReadPermissionPinned(defs []schema.TableDefinition, result *AssertionResult) {
	log.Println("   📋 Checking _System_ExternalDeliveryAttempt requiredReadPermission pin...")

	for _, def := range defs {
		if def.TableName != "_System_ExternalDeliveryAttempt" {
			continue
		}
		if def.RequiredReadPermission != externalDeliveryAttemptRequiredReadPermission {
			result.Violations = append(result.Violations, AssertionViolation{
				Category: "AuditReadGate",
				Severity: constants.SeverityError,
				Object:   "_System_ExternalDeliveryAttempt",
				Description: fmt.Sprintf(
					`_System_ExternalDeliveryAttempt.requiredReadPermission must be %q (got %q); `+
						`frozen outbound payloads can contain L3 record data + delivery secrets and are exposed only through the record-scoped delivery admin surface`,
					externalDeliveryAttemptRequiredReadPermission, def.RequiredReadPermission,
				),
			})
		}
		return
	}
	result.Violations = append(result.Violations, AssertionViolation{
		Category:    "AuditReadGate",
		Severity:    constants.SeverityError,
		Object:      "_System_ExternalDeliveryAttempt",
		Description: "_System_ExternalDeliveryAttempt entry missing from system_tables.json; cannot verify delivery read gate",
	})
}

// assertExternalDeliveryAttemptReadPermissionLiveDB enforces the same
// invariant as assertExternalDeliveryAttemptReadPermissionPinned, but against
// the live _System_Object row that runtime permission checks actually read.
func assertExternalDeliveryAttemptReadPermissionLiveDB(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking _System_ExternalDeliveryAttempt RequiredReadPermission live DB row...")

	var liveValue sql.NullString
	err := db.QueryRow(
		fmt.Sprintf(
			"SELECT `%s` FROM `%s` WHERE `%s` = ? LIMIT 1",
			constants.FieldSysObject_RequiredReadPermission,
			constants.TableObject,
			constants.FieldAPIName,
		),
		"_System_ExternalDeliveryAttempt",
	).Scan(&liveValue)

	if err == sql.ErrNoRows {
		log.Println("   ℹ️  _System_ExternalDeliveryAttempt row not found in DB (fresh DB?); skipping live-row RRP check")
		return
	}
	if err != nil {
		log.Printf("   ⚠️  Could not query _System_Object for _System_ExternalDeliveryAttempt: %v", err)
		return
	}

	live := liveValue.String
	if live != externalDeliveryAttemptRequiredReadPermission {
		result.Violations = append(result.Violations, AssertionViolation{
			Category: "AuditReadGate",
			Severity: constants.SeverityError,
			Object:   "_System_ExternalDeliveryAttempt",
			Description: fmt.Sprintf(
				`_System_Object live row for _System_ExternalDeliveryAttempt has RequiredReadPermission=%q (want %q); `+
					`the DB has drifted from system_tables.json — outbound delivery payload gate is using the wrong value. `+
					`Frozen outbound payloads can contain L3 record data + delivery secrets and are exposed only through the record-scoped delivery admin surface. `+
					`Run setup apply to reconcile, or restore the value via DB if reconciliation has been deliberately disabled.`,
				live, externalDeliveryAttemptRequiredReadPermission,
			),
		})
	}
}

// assertExternalDeliveryAttemptWritePermissionPinned is a parse-time invariant
// on the embedded system_tables.json definitions (no DB read).
//
// _System_ExternalDeliveryAttempt must be "BLOCKED" for generic CRUD writes.
// Only the external-delivery service may insert original or resend rows via
// direct SQL; allowing generic writes would let API callers forge or mutate
// operator evidence.
func assertExternalDeliveryAttemptWritePermissionPinned(defs []schema.TableDefinition, result *AssertionResult) {
	log.Println("   📋 Checking _System_ExternalDeliveryAttempt requiredWritePermission pin...")

	for _, def := range defs {
		if def.TableName != "_System_ExternalDeliveryAttempt" {
			continue
		}
		if def.RequiredWritePermission != externalDeliveryAttemptRequiredWritePermission {
			result.Violations = append(result.Violations, AssertionViolation{
				Category: "AuditWriteGate",
				Severity: constants.SeverityError,
				Object:   "_System_ExternalDeliveryAttempt",
				Description: fmt.Sprintf(
					`_System_ExternalDeliveryAttempt.requiredWritePermission must be %q (got %q); `+
						`requiredWritePermission must stay BLOCKED so generic CRUD cannot forge or rewrite delivery evidence`,
					externalDeliveryAttemptRequiredWritePermission, def.RequiredWritePermission,
				),
			})
		}
		return
	}
	result.Violations = append(result.Violations, AssertionViolation{
		Category:    "AuditWriteGate",
		Severity:    constants.SeverityError,
		Object:      "_System_ExternalDeliveryAttempt",
		Description: "_System_ExternalDeliveryAttempt entry missing from system_tables.json; cannot verify delivery write gate",
	})
}

// assertExternalDeliveryAttemptWritePermissionLiveDB enforces the same
// invariant as assertExternalDeliveryAttemptWritePermissionPinned, but against
// the live _System_Object row.
func assertExternalDeliveryAttemptWritePermissionLiveDB(db *sql.DB, result *AssertionResult) {
	log.Println("   📋 Checking _System_ExternalDeliveryAttempt RequiredWritePermission live DB row...")

	var liveValue sql.NullString
	err := db.QueryRow(
		fmt.Sprintf(
			"SELECT `%s` FROM `%s` WHERE `%s` = ? LIMIT 1",
			constants.FieldSysObject_RequiredWritePermission,
			constants.TableObject,
			constants.FieldAPIName,
		),
		"_System_ExternalDeliveryAttempt",
	).Scan(&liveValue)

	if err == sql.ErrNoRows {
		log.Println("   ℹ️  _System_ExternalDeliveryAttempt row not found in DB (fresh DB?); skipping live-row WRP check")
		return
	}
	if err != nil {
		log.Printf("   ⚠️  Could not query _System_Object WRP for _System_ExternalDeliveryAttempt: %v", err)
		return
	}

	live := liveValue.String
	if live != externalDeliveryAttemptRequiredWritePermission {
		result.Violations = append(result.Violations, AssertionViolation{
			Category: "AuditWriteGate",
			Severity: constants.SeverityError,
			Object:   "_System_ExternalDeliveryAttempt",
			Description: fmt.Sprintf(
				`_System_Object live row for _System_ExternalDeliveryAttempt has RequiredWritePermission=%q (want %q); `+
					`the DB has drifted from system_tables.json — generic CRUD write gate for delivery evidence is using the wrong value. `+
					`Run setup apply to reconcile, or restore the value via DB if reconciliation has been deliberately disabled.`,
				live, externalDeliveryAttemptRequiredWritePermission,
			),
		})
	}
}

// assertWritableSystemCoreTablesHaveKeyPrefix enforces two parse-time invariants
// on the embedded system_tables.json definitions. It is a pure in-memory check
// (no DB read); defs is the caller's already-loaded slice.
//
// Uniqueness (SeverityError): no two entries in system_tables.json may share
// the same non-empty keyPrefix. Duplicate prefixes cause record-ID namespace
// collisions and are irrecoverable without a code change. Server refuses to
// start in strict mode when this violation is present.
//
// Coverage (SeverityWarning): every system_core entry with a named
// requiredWritePermission (non-empty, not "BLOCKED") must declare a non-empty
// keyPrefix. Missing keyPrefixes cause HTTP 500 on record creates for the
// affected table. Violations are logged as warnings — not errors — because
// filling in the remaining ~14 naked entries is progressive work tracked by
// follow-up issues. Server starts but operators are alerted.
func assertWritableSystemCoreTablesHaveKeyPrefix(defs []schema.TableDefinition, result *AssertionResult) {
	log.Println("   📋 Checking system_core keyPrefix uniqueness + coverage...")

	// Uniqueness: scan all entries, flag any duplicate non-empty keyPrefix.
	seen := make(map[string]string) // keyPrefix → first tableName
	for _, def := range defs {
		if def.KeyPrefix == "" {
			continue
		}
		if first, dup := seen[def.KeyPrefix]; dup {
			result.Violations = append(result.Violations, AssertionViolation{
				Category:    "KeyPrefixUniqueness",
				Severity:    constants.SeverityError,
				Object:      def.TableName,
				Description: fmt.Sprintf("duplicate keyPrefix %q: declared by %q and %q; record-ID namespace collision is irrecoverable", def.KeyPrefix, first, def.TableName),
			})
		} else {
			seen[def.KeyPrefix] = def.TableName
		}
	}

	// Coverage: warn for writable system_core entries without a keyPrefix.
	// "Writable" = tableType is system_core AND requiredWritePermission is
	// a named permission (non-empty string that is not "BLOCKED").
	for _, def := range defs {
		if def.TableType != string(constants.TableTypeSystemCore) {
			continue
		}
		if def.RequiredWritePermission == "" || def.RequiredWritePermission == "BLOCKED" {
			continue
		}
		if def.KeyPrefix == "" {
			result.Violations = append(result.Violations, AssertionViolation{
				Category:    "KeyPrefixCoverage",
				Severity:    constants.SeverityWarning,
				Object:      def.TableName,
				Description: fmt.Sprintf("writable system_core table %q (requiredWritePermission=%q) has no keyPrefix; flows and REST creates targeting this table will return HTTP 500", def.TableName, def.RequiredWritePermission),
			})
		}
	}
}