package packs

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sort"
	"strings"

	"github.com/nexuscrm/db/pkg/schemashape"
	domainschema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
)

type ownerlessObjectSpecCompatibilityError struct {
	err error
}

func (e *ownerlessObjectSpecCompatibilityError) Error() string {
	return e.err.Error()
}

func (e *ownerlessObjectSpecCompatibilityError) Unwrap() error {
	return e.err
}

func newOwnerlessObjectSpecCompatibilityError(format string, args ...any) error {
	return &ownerlessObjectSpecCompatibilityError{err: fmt.Errorf(format, args...)}
}

func isOwnerlessObjectSpecCompatibilityError(err error) bool {
	var target *ownerlessObjectSpecCompatibilityError
	return errors.As(err, &target)
}

func guardSourceBackedObjectSpecs(ctx context.Context, db *sql.DB, packName string, defs []domainschema.TableDefinition, allowed map[string]bool, requiredFields ...objectSpecRequiredFieldIndex) error {
	return guardSourceBackedObjectSpecsWithComponentKindRegistrationSet(ctx, db, packName, defs, allowed, nil, requiredFields...)
}

func guardSourceBackedObjectSpecsWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, packName string, defs []domainschema.TableDefinition, allowed map[string]bool, registrations *ComponentKindRegistrationSet, requiredFields ...objectSpecRequiredFieldIndex) error {
	compatibleExistingDefs, err := inspectSourceBackedObjectSpecsWithComponentKindRegistrationSet(
		ctx, db, packName, defs, allowed, registrations, requiredFields...,
	)
	if err != nil {
		return err
	}
	return adoptLegacyObjectSpecFieldOwnership(ctx, db, packName, compatibleExistingDefs)
}

// ValidateStoppedSourceSetObjectSpecCompatibility establishes one pack-wide,
// read-only compatibility barrier across every object_spec component. It is
// invoked from source-pack preflight before runtime registration or lifecycle
// mutation; InstallPlan invokes the same barrier again immediately before
// component writes.
func ValidateStoppedSourceSetObjectSpecCompatibility(ctx context.Context, deps ApplyDeps, pack *SourceBackedPack) error {
	if pack == nil {
		return fmt.Errorf("stopped source-set object-spec compatibility requires a pack")
	}
	if !offlinecandidateauthority.StoppedSourceSetApplyOnly(ctx, pack.Name()) {
		return fmt.Errorf("stopped source-set object-spec compatibility requires apply-only authority for pack %s", pack.Name())
	}
	plan, err := pack.InstallPlan()
	if err != nil {
		return err
	}
	return plan.validateStoppedSourceSetObjectSpecCompatibility(ctx, deps)
}

func (p InstallPlan) validateStoppedSourceSetObjectSpecCompatibility(ctx context.Context, deps ApplyDeps) error {
	preflightCtx, err := preflightMetadataTransactionExecutor(ctx, deps, p.packName)
	if err != nil {
		return err
	}
	seenGroups := make(map[string]struct{})
	var compatibilityErrs []error
	for _, phase := range p.phases {
		for _, rawComponent := range p.steps[phase] {
			component, ok := rawComponent.(*sourceBackedObjectSpecComponent)
			if !ok {
				continue
			}
			groupKey := componentInstallApplyGroupKey(component)
			if _, seen := seenGroups[groupKey]; seen {
				continue
			}
			seenGroups[groupKey] = struct{}{}
			defs, _, err := component.applyDefinitions(preflightCtx, deps)
			if err != nil {
				return fmt.Errorf("object_spec %s: %w", component.BusinessKey(), err)
			}
			_, err = inspectSourceBackedObjectSpecsWithComponentKindRegistrationSet(
				preflightCtx,
				deps.DB,
				component.packName,
				defs,
				component.allowOwnerlessObjectAPI,
				component.registrations,
				newObjectSpecRequiredFieldIndex(component.specs),
			)
			if err == nil {
				continue
			}
			wrapped := fmt.Errorf("object_spec %s: %w", component.BusinessKey(), err)
			if !isOwnerlessObjectSpecCompatibilityError(err) {
				return wrapped
			}
			compatibilityErrs = append(compatibilityErrs, wrapped)
		}
	}
	return errors.Join(compatibilityErrs...)
}

// inspectSourceBackedObjectSpecsWithComponentKindRegistrationSet is the
// read-only half of ownerless object-spec admission. Pack-level stopped apply
// invokes it for every object_spec component before any component can adopt
// ownership; the component guard invokes it again immediately before apply as
// defense in depth.
func inspectSourceBackedObjectSpecsWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, packName string, defs []domainschema.TableDefinition, allowed map[string]bool, registrations *ComponentKindRegistrationSet, requiredFields ...objectSpecRequiredFieldIndex) ([]domainschema.TableDefinition, error) {
	requiredFieldIndex := objectSpecRequiredFieldIndexFromArgs(requiredFields)
	physicalTables, err := loadPhysicalTableSet(db)
	if err != nil {
		return nil, fmt.Errorf("load physical tables: %w", err)
	}
	physicalColumns, err := schemaProvisioner.LoadPhysicalColumnShapes(db)
	if err != nil {
		return nil, fmt.Errorf("load physical columns: %w", err)
	}
	tableRegistry, err := loadPackTableRegistry(db)
	if err != nil {
		return nil, fmt.Errorf("load table registry: %w", err)
	}
	objectRegistry, err := loadPackObjectRegistry(db)
	if err != nil {
		return nil, fmt.Errorf("load object registry: %w", err)
	}
	fieldRegistry, err := loadPackFieldRegistry(db)
	if err != nil {
		return nil, fmt.Errorf("load field registry: %w", err)
	}

	compatibleExistingDefs := make([]domainschema.TableDefinition, 0, len(defs))
	var compatibilityErrs []error
	for _, def := range defs {
		key := strings.ToLower(def.TableName)
		hasSurface := physicalTables[key] || tableRegistry[key].tableType != "" || objectRegistry[key].tableType != "" || len(fieldRegistry[key]) > 0
		if !hasSurface {
			continue
		}
		if !allowed[key] {
			compatibilityErrs = append(compatibilityErrs, newOwnerlessObjectSpecCompatibilityError(
				"ownerless object/table/field surface %s already exists and is not same-pack retry-compatible for pack %s",
				def.TableName,
				packName,
			))
			continue
		}
		if err := verifyOwnerlessObjectSpecPartialCompatibilityWithComponentKindRegistrationSet(ctx, db, packName, def, physicalTables, physicalColumns, tableRegistry, objectRegistry, fieldRegistry, registrations, requiredFieldIndex); err != nil {
			wrapped := fmt.Errorf("ownerless %s is not compatible with pack %s retry: %w", def.TableName, packName, err)
			if !isOwnerlessObjectSpecCompatibilityError(err) {
				return nil, wrapped
			}
			compatibilityErrs = append(compatibilityErrs, wrapped)
			continue
		}
		compatibleExistingDefs = append(compatibleExistingDefs, def)
	}
	if len(compatibilityErrs) > 0 {
		return nil, errors.Join(compatibilityErrs...)
	}
	return compatibleExistingDefs, nil
}

func adoptLegacyObjectSpecFieldOwnership(ctx context.Context, db *sql.DB, packName string, defs []domainschema.TableDefinition) error {
	for _, def := range defs {
		expectedFields := map[string]bool{}
		expectedFieldNames := make([]string, 0, len(def.Columns))
		for _, col := range def.Columns {
			if packColumnRequiresFieldMetadata(col.Name) {
				expectedFields[strings.ToLower(col.Name)] = true
				expectedFieldNames = append(expectedFieldNames, col.Name)
			}
		}
		if len(expectedFieldNames) > 0 {
			args := make([]any, 0, 5+len(expectedFieldNames))
			args = append(args,
				constants.FieldOwnerKindPack,
				packName,
				def.TableName,
			)
			inPlaceholders := make([]string, 0, len(expectedFieldNames))
			for _, fieldName := range expectedFieldNames {
				inPlaceholders = append(inPlaceholders, "?")
				args = append(args, fieldName)
			}
			args = append(args,
				constants.FieldOwnerKindRuntime,
				constants.FieldOwnerKindPack,
				packName,
			)
			if _, err := db.ExecContext(ctx, fmt.Sprintf(`
UPDATE _System_Field f
JOIN _System_Object o ON o.Id = f.ObjectId
   SET f.OwnerKind = ?, f.OwnerPack = ?, f.LastModifiedDate = CURRENT_TIMESTAMP
 WHERE COALESCE(o.IsDeleted, 0) = 0
   AND COALESCE(f.IsDeleted, 0) = 0
   AND o.ApiName = ?
   AND f.ApiName IN (%s)
   AND (COALESCE(f.OwnerKind, '') = '' OR f.OwnerKind = ? OR (f.OwnerKind = ? AND f.OwnerPack = ?))`,
				strings.Join(inPlaceholders, ",")), args...); err != nil {
				return fmt.Errorf("adopt pack field owners %s: %w", def.TableName, err)
			}
		}
		rows, err := db.QueryContext(ctx, `
SELECT f.Id, f.ApiName
  FROM _System_Field f
  JOIN _System_Object o ON o.Id = f.ObjectId
 WHERE COALESCE(o.IsDeleted, 0) = 0
   AND COALESCE(f.IsDeleted, 0) = 0
   AND o.ApiName = ?
   AND (COALESCE(f.OwnerKind, '') = '' OR f.OwnerKind = ?)`,
			def.TableName,
			constants.FieldOwnerKindRuntime,
		)
		if err != nil {
			return fmt.Errorf("load legacy ownerless fields for %s: %w", def.TableName, err)
		}
		var adminFieldIDs []string
		for rows.Next() {
			var fieldID, fieldName string
			if err := rows.Scan(&fieldID, &fieldName); err != nil {
				rows.Close()
				return fmt.Errorf("scan legacy ownerless field for %s: %w", def.TableName, err)
			}
			if expectedFields[strings.ToLower(fieldName)] {
				continue
			}
			if isCustomFieldAPIName(fieldName) {
				adminFieldIDs = append(adminFieldIDs, fieldID)
			}
		}
		if err := rows.Err(); err != nil {
			rows.Close()
			return fmt.Errorf("iterate legacy ownerless fields for %s: %w", def.TableName, err)
		}
		rows.Close()
		for _, fieldID := range adminFieldIDs {
			if _, err := db.ExecContext(ctx, `
UPDATE _System_Field
   SET OwnerKind = ?, OwnerPack = NULL, LastModifiedDate = CURRENT_TIMESTAMP
 WHERE Id = ?
   AND (COALESCE(OwnerKind, '') = '' OR OwnerKind = ?)`,
				constants.FieldOwnerKindAdmin,
				fieldID,
				constants.FieldOwnerKindRuntime,
			); err != nil {
				return fmt.Errorf("adopt admin field owner %s: %w", fieldID, err)
			}
		}
	}

	return nil
}

func verifyOwnerlessObjectSpecPartialCompatibility(
	ctx context.Context,
	db *sql.DB,
	packName string,
	def domainschema.TableDefinition,
	physicalTables map[string]bool,
	physicalColumns map[string]map[string]schemashape.PhysicalColumnShape,
	tableRegistry map[string]packTableRegistryRow,
	objectRegistry map[string]packObjectRegistryRow,
	fieldRegistry map[string]map[string]packFieldRegistryRow,
	requiredFields ...objectSpecRequiredFieldIndex,
) error {
	return verifyOwnerlessObjectSpecPartialCompatibilityWithComponentKindRegistrationSet(
		ctx, db, packName, def, physicalTables, physicalColumns,
		tableRegistry, objectRegistry, fieldRegistry, nil, requiredFields...,
	)
}

func verifyOwnerlessObjectSpecPartialCompatibilityWithComponentKindRegistrationSet(
	ctx context.Context,
	db *sql.DB,
	packName string,
	def domainschema.TableDefinition,
	physicalTables map[string]bool,
	physicalColumns map[string]map[string]schemashape.PhysicalColumnShape,
	tableRegistry map[string]packTableRegistryRow,
	objectRegistry map[string]packObjectRegistryRow,
	fieldRegistry map[string]map[string]packFieldRegistryRow,
	registrations *ComponentKindRegistrationSet,
	requiredFields ...objectSpecRequiredFieldIndex,
) error {
	requiredFieldIndex := objectSpecRequiredFieldIndexFromArgs(requiredFields)
	var compatibilityErrs []error
	key := strings.ToLower(def.TableName)
	if tableRow, ok := tableRegistry[key]; ok {
		if tableRow.deleted {
			compatibilityErrs = append(compatibilityErrs, newOwnerlessObjectSpecCompatibilityError("_System_Table row is deleted"))
		}
		if !tableRow.managed {
			compatibilityErrs = append(compatibilityErrs, newOwnerlessObjectSpecCompatibilityError("_System_Table IsManaged = 0, want 1"))
		}
		if tableRow.tableType != def.TableType {
			compatibilityErrs = append(compatibilityErrs, newOwnerlessObjectSpecCompatibilityError("_System_Table TableType = %q, want %q", tableRow.tableType, def.TableType))
		}
		if tableRow.category != def.Category {
			compatibilityErrs = append(compatibilityErrs, newOwnerlessObjectSpecCompatibilityError("_System_Table Category = %q, want %q", tableRow.category, def.Category))
		}
	}
	if objectRow, ok := objectRegistry[key]; ok {
		if objectRow.deleted {
			compatibilityErrs = append(compatibilityErrs, newOwnerlessObjectSpecCompatibilityError("_System_Object row is deleted"))
		}
		if objectRow.tableType != def.TableType {
			compatibilityErrs = append(compatibilityErrs, newOwnerlessObjectSpecCompatibilityError("_System_Object TableType = %q, want %q", objectRow.tableType, def.TableType))
		}
	}
	if physicalTables[key] {
		actualColumns := physicalColumns[key]
		for _, col := range def.Columns {
			colKey := strings.ToLower(col.Name)
			if actual, ok := actualColumns[colKey]; ok {
				compatible, err := packObjectSpecColumnShapeRetryCompatible(ctx, db, def.TableName, actual, col, requiredFieldIndex)
				if err != nil {
					if !isOwnerlessObjectSpecCompatibilityError(err) {
						return err
					}
					compatibilityErrs = append(compatibilityErrs, err)
					continue
				}
				if compatible {
					continue
				}
				compatibilityErrs = append(compatibilityErrs, newOwnerlessObjectSpecCompatibilityError("physical column %s.%s shape = %s nullable=%v default=%s, want %s nullable=%v default=%s",
					def.TableName,
					col.Name,
					actual.ColumnType,
					actual.Nullable,
					formatPackColumnDefault(actual.DefaultVal.Valid, actual.DefaultVal.String),
					schemaProvisioner.DesiredColumnType(col),
					col.Nullable,
					formatPackColumnDefault(col.Default != "", col.Default),
				))
			}
		}
	}
	expectedFields := map[string]bool{}
	for _, col := range def.Columns {
		if packColumnRequiresFieldMetadata(col.Name) {
			expectedFields[strings.ToLower(col.Name)] = true
		}
	}
	expectedFieldNames := make([]string, 0, len(expectedFields))
	for fieldName := range expectedFields {
		expectedFieldNames = append(expectedFieldNames, fieldName)
	}
	sort.Strings(expectedFieldNames)
	for _, fieldName := range expectedFieldNames {
		if row, ok := fieldRegistry[key][fieldName]; ok && !row.isOwnedByPack(packName) && !row.isOwnerlessOrRuntime() {
			compatibilityErrs = append(compatibilityErrs, newOwnerlessObjectSpecCompatibilityError("_System_Field row %s.%s owner = %s/%s, want pack/%s or legacy ownerless/runtime", def.TableName, fieldName, row.ownerKind, row.ownerPack, packName))
		}
	}
	var crossPackVerifier *crossPackFieldVerifier
	loadCrossPackVerifier := func() (*crossPackFieldVerifier, error) {
		if crossPackVerifier != nil {
			return crossPackVerifier, nil
		}
		verifier, err := buildCrossPackFieldVerifierWithComponentKindRegistrationSet(ctx, db, packName, []domainschema.TableDefinition{def}, fieldRegistry, registrations)
		if err != nil {
			return nil, err
		}
		crossPackVerifier = verifier
		return crossPackVerifier, nil
	}
	registeredFieldNames := make([]string, 0, len(fieldRegistry[key]))
	for fieldName := range fieldRegistry[key] {
		registeredFieldNames = append(registeredFieldNames, fieldName)
	}
	sort.Strings(registeredFieldNames)
	for _, fieldName := range registeredFieldNames {
		if !expectedFields[fieldName] {
			row := fieldRegistry[key][fieldName]
			displayFieldAPIName := strings.TrimSpace(row.fieldAPIName)
			if displayFieldAPIName == "" {
				displayFieldAPIName = fieldName
			}
			if row.isAdminOwned() || (row.isOwnerlessOrRuntime() && isCustomFieldAPIName(fieldName)) {
				continue
			}
			if row.isOtherPackManagedField(packName) {
				verifier, err := loadCrossPackVerifier()
				if err != nil {
					return fmt.Errorf("build cross-pack field verifier for %s: %w", def.TableName, err)
				}
				if verifier.allowsDeclaredPeerPackField(row, def.TableName, fieldName, packName) {
					continue
				}
			}
			compatibilityErrs = append(compatibilityErrs, newOwnerlessObjectSpecCompatibilityError("_System_Field row %s.%s is not declared by the pack, owned by admin, legacy custom field residue, or declared peer-pack field", def.TableName, displayFieldAPIName))
		}
	}
	return errors.Join(compatibilityErrs...)
}