package packs

import (
	"context"
	"regexp"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	domainschema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/formulastorage"
	"github.com/nexuscrm/shared/pkg/models"
)

func TestBuildObjectMetadataPropagatesReadSource(t *testing.T) {
	specs := []ObjectSpec{{
		APIName:      "RecordType",
		Label:        "RecordType",
		TableType:    "standard_object",
		Category:     "system",
		SharingModel: "PublicRead",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
	}}

	objects := buildObjectMetadataWithResolver(specs, objectSpecFieldResolver{})
	if len(objects) != 1 {
		t.Fatalf("buildObjectMetadataWithResolver returned %d objects, want 1", len(objects))
	}
	if objects[0].ReadSource == nil {
		t.Fatal("ReadSource = nil, want propagated config")
	}
	if objects[0].ReadSource.Kind != "physical_table" {
		t.Fatalf("ReadSource.Kind = %q, want %q", objects[0].ReadSource.Kind, "physical_table")
	}
	if objects[0].ReadSource.PhysicalTable != "_System_RecordType" {
		t.Fatalf("ReadSource.PhysicalTable = %q, want %q", objects[0].ReadSource.PhysicalTable, "_System_RecordType")
	}

	// Ensure the metadata builder clones the config instead of reusing spec memory.
	specs[0].ReadSource.PhysicalTable = "Mutated"
	if objects[0].ReadSource.PhysicalTable != "_System_RecordType" {
		t.Fatalf("ReadSource.PhysicalTable mutated to %q, want isolated copy", objects[0].ReadSource.PhysicalTable)
	}
}

func TestBuildObjectMetadataPropagatesFormulaStorageMode(t *testing.T) {
	specs := []ObjectSpec{{
		APIName: "WorkItem",
		Fields: []FieldSpec{{
			APIName:            "NameStored",
			Label:              "Name Stored",
			Type:               "formula",
			Formula:            "Name",
			FormulaMode:        "active",
			FormulaStorageMode: "write_materialized",
			ReturnType:         "string",
		}},
	}}

	objects := buildObjectMetadataWithResolver(specs, objectSpecFieldResolver{})
	if len(objects) != 1 {
		t.Fatalf("buildObjectMetadataWithResolver returned %d objects, want 1", len(objects))
	}
	if len(objects[0].Fields) != 1 {
		t.Fatalf("buildObjectMetadataWithResolver fields = %d, want 1", len(objects[0].Fields))
	}
	if objects[0].Fields[0].FormulaStorageMode == nil {
		t.Fatal("FormulaStorageMode = nil, want propagated write_materialized")
	}
	if got := *objects[0].Fields[0].FormulaStorageMode; got != "write_materialized" {
		t.Fatalf("FormulaStorageMode = %q, want write_materialized", got)
	}
}

func TestStandardObjectSpecRecordTypeOmitsUnsupportedReadSourceFields(t *testing.T) {
	specs, err := parseStandardObjectsSpec()
	if err != nil {
		t.Fatalf("parseStandardObjectsSpec() error = %v", err)
	}

	spec := findObjectSpec(t, specs, "RecordType")
	// NamespacePrefix is backed by _System_RecordType (system_tables.json) and must be declared.
	// SobjectType is the discriminator column and must remain in the read contract.
	for _, required := range []string{"SobjectType", "NamespacePrefix"} {
		if findFieldSpec(t, spec, required).APIName != required {
			t.Fatalf("RecordType field %q must be declared in the read contract (backed in _System_RecordType)", required)
		}
	}
}

func TestConvertSpecToTableDefinitionsIncludesReadSourceObjectInMetadataDefs(t *testing.T) {
	sm := metadata.NewSchemaManager(persistence.NewSchemaRepository(nil, nil))
	specs := []ObjectSpec{{
		APIName:      "RecordType",
		Label:        "RecordType",
		TableType:    "standard_object",
		Category:     "system",
		SharingModel: "PublicRead",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
	}}

	defs, err := convertSpecToTableDefinitionsWithResolver(specs, sm, objectSpecFieldResolver{})
	if err != nil {
		t.Fatalf("convertSpecToTableDefinitionsWithResolver() error = %v", err)
	}
	if len(defs) != 1 {
		t.Fatalf("convertSpecToTableDefinitionsWithResolver() returned %d defs, want 1 metadata definition", len(defs))
	}
	if defs[0].ReadSource == nil {
		t.Fatal("defs[0].ReadSource = nil, want preserved readSource contract")
	}
	if defs[0].ReadSource.PhysicalTable != "_System_RecordType" {
		t.Fatalf("defs[0].ReadSource.PhysicalTable = %q, want %q", defs[0].ReadSource.PhysicalTable, "_System_RecordType")
	}
}

func TestPhysicalTableDefinitionsSkipsReadSourceObject(t *testing.T) {
	defs := []domainschema.TableDefinition{{
		TableName: "RecordType",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
	}}

	physical := make([]domainschema.TableDefinition, 0, len(defs))
	for _, def := range defs {
		if def.ReadSource != nil {
			continue
		}
		physical = append(physical, def)
	}
	if len(physical) != 0 {
		t.Fatalf("physical defs = %d, want 0", len(physical))
	}
}

func TestPhysicalTableDefinitionsSkipViewOnlyActiveFormulaColumns(t *testing.T) {
	defs := []domainschema.TableDefinition{{
		TableName: "WorkItem",
		Columns: []domainschema.ColumnDefinition{
			{Name: "Id", Type: "VARCHAR(18)", LogicalType: "string"},
			{
				Name:        "NameEcho",
				Type:        "VARCHAR(255)",
				LogicalType: "formula",
				Formula:     "Name",
				FormulaMode: "active",
				ReturnType:  "string",
				Nullable:    true,
			},
			{
				Name:               "NameStored",
				Type:               "VARCHAR(255)",
				LogicalType:        "formula",
				Formula:            "Name",
				FormulaMode:        "active",
				FormulaStorageMode: "write_materialized",
				ReturnType:         "string",
				Nullable:           true,
			},
			{
				Name:        "NameFrozen",
				Type:        "VARCHAR(255)",
				LogicalType: "formula",
				Formula:     "Name",
				FormulaMode: "frozen",
				ReturnType:  "string",
				Nullable:    true,
			},
		},
	}}

	physical := make([]domainschema.TableDefinition, 0, len(defs))
	for _, def := range defs {
		if def.ReadSource != nil {
			continue
		}
		physicalCols := make([]domainschema.ColumnDefinition, 0, len(def.Columns))
		for _, col := range def.Columns {
			resolved, err := formulastorage.Resolve(formulastorage.SpecFromColumn(col))
			if !resolved.RequiresPhysicalColumn() {
				continue
			}
			physicalCols = append(physicalCols, col)
			_ = err
		}
		def.Columns = physicalCols
		physical = append(physical, def)
	}
	if len(physical) != 1 {
		t.Fatalf("physical defs = %d, want 1", len(physical))
	}
	cols := physical[0].Columns
	if readSourceHasColumn(cols, "NameEcho") {
		t.Fatalf("physical columns = %#v, want view-only active formula NameEcho excluded", cols)
	}
	for _, want := range []string{"Id", "NameStored", "NameFrozen"} {
		if !readSourceHasColumn(cols, want) {
			t.Fatalf("physical columns = %#v, missing %s", cols, want)
		}
	}
}

func TestPhysicalTableDefinitionsUseCanonicalFallbackWhenStorageModeIsInvalid(t *testing.T) {
	defs := []domainschema.TableDefinition{{
		TableName: "WorkItem",
		Columns: []domainschema.ColumnDefinition{
			{Name: "Id", Type: "VARCHAR(18)", LogicalType: "string"},
			{
				Name:               "NameEcho",
				Type:               "VARCHAR(255)",
				LogicalType:        "formula",
				Formula:            "Name",
				FormulaMode:        "active",
				FormulaStorageMode: "WRITE_MATERIALIZED",
				ReturnType:         "string",
				Nullable:           true,
			},
		},
	}}

	physical := make([]domainschema.TableDefinition, 0, len(defs))
	for _, def := range defs {
		if def.ReadSource != nil {
			continue
		}
		physicalCols := make([]domainschema.ColumnDefinition, 0, len(def.Columns))
		for _, col := range def.Columns {
			resolved, err := formulastorage.Resolve(formulastorage.SpecFromColumn(col))
			if !resolved.RequiresPhysicalColumn() {
				continue
			}
			physicalCols = append(physicalCols, col)
			_ = err
		}
		def.Columns = physicalCols
		physical = append(physical, def)
	}
	if len(physical) != 1 {
		t.Fatalf("physical defs = %d, want 1", len(physical))
	}
	if readSourceHasColumn(physical[0].Columns, "NameEcho") {
		t.Fatalf("physical columns = %#v, want invalid active formulaStorageMode to fall back to view-only and exclude NameEcho", physical[0].Columns)
	}
}

func readSourceHasColumn(cols []domainschema.ColumnDefinition, want string) bool {
	for _, col := range cols {
		if col.Name == want {
			return true
		}
	}
	return false
}

func TestValidateSpecReadSourcesRejectsMalformedDeclaration(t *testing.T) {
	specs := []ObjectSpec{{
		APIName:      "RecordType",
		Label:        "RecordType",
		TableType:    "standard_object",
		Category:     "system",
		SharingModel: "PublicRead",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "view",
			PhysicalTable: "_System_RecordType",
		},
	}}

	if err := validateSpecReadSources(specs); err == nil {
		t.Fatal("validateSpecReadSources() = nil, want invalid declaration failure")
	}
}

func TestValidateSpecReadSourcesRejectsUnsupportedObjectShape(t *testing.T) {
	specs := []ObjectSpec{{
		APIName:      "Widget__c",
		Label:        "Widget",
		TableType:    "custom_object",
		Category:     "crm",
		SharingModel: "Private",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
	}}

	err := validateSpecReadSources(specs)
	if err == nil {
		t.Fatal("validateSpecReadSources() = nil, want unsupported readSource authoring failure")
	}
	if !strings.Contains(err.Error(), "not yet supported in the current pack contract") {
		t.Fatalf("validateSpecReadSources() error = %v, want deferred-contract detail", err)
	}
}

func TestValidateSpecReadSourcesRejectsRecordTypeAuthoringForNow(t *testing.T) {
	specs := []ObjectSpec{{
		APIName:      "RecordType",
		Label:        "RecordType",
		TableType:    "standard_object",
		Category:     "system",
		SharingModel: "PublicRead",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
	}}

	err := validateSpecReadSources(specs)
	if err == nil {
		t.Fatal("validateSpecReadSources() = nil, want readSource contract rejection")
	}
	if !strings.Contains(err.Error(), "not yet supported in the current pack contract") {
		t.Fatalf("validateSpecReadSources() error = %v, want deferred-contract detail", err)
	}
}

func TestValidateSpecReadSourcesForApplyAllowsPlatformCoreRecordType(t *testing.T) {
	specs := []ObjectSpec{{
		APIName:      "RecordType",
		Label:        "RecordType",
		TableType:    "standard_object",
		Category:     "system",
		SharingModel: "PublicRead",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
	}}

	if err := validateSpecReadSourcesForApply("platform-core", specs); err != nil {
		t.Fatalf("validateSpecReadSourcesForApply(platform-core) = %v, want nil", err)
	}
}

func TestValidateSpecReadSourcesForApplyRejectsNonCanonicalPlatformCoreRecordTypeSourceTable(t *testing.T) {
	specs := []ObjectSpec{{
		APIName:      "RecordType",
		Label:        "RecordType",
		TableType:    "standard_object",
		Category:     "system",
		SharingModel: "PublicRead",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "RecordTypeShadow",
		},
	}}

	err := validateSpecReadSourcesForApply("platform-core", specs)
	if err == nil {
		t.Fatal("validateSpecReadSourcesForApply(platform-core) = nil, want canonical physical-table failure")
	}
	if !strings.Contains(err.Error(), constants.TableRecordType) {
		t.Fatalf("validateSpecReadSourcesForApply(platform-core) error = %v, want canonical table detail", err)
	}
}

func TestValidateReadSourceLegacySurfaceCompatibilityFailsOnMissingBackedColumnBeforePublish(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	defs := []domainschema.TableDefinition{{
		TableName: "Widget",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
		Columns: []domainschema.ColumnDefinition{
			{Name: constants.FieldCreatedByID},
		},
	}}
	fields := []metadata.FieldWithContext{
		{TableName: "Widget", Field: &models.FieldMetadata{APIName: constants.FieldCreatedByID}},
	}

	mock.ExpectQuery("SELECT TABLE_NAME FROM INFORMATION_SCHEMA\\.TABLES").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME"}).AddRow("_System_RecordType"))
	mock.ExpectQuery("SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLLATION_NAME FROM INFORMATION_SCHEMA\\.COLUMNS").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME", "COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE", "COLUMN_DEFAULT", "COLLATION_NAME"}).
			AddRow("_System_RecordType", "Id", "varchar(18)", "NO", nil, nil))

	err = validateReadSourceLegacySurfaceCompatibility(context.Background(), db, "crm", defs, fields)
	if err == nil {
		t.Fatal("validateReadSourceLegacySurfaceCompatibility() = nil, want backing-column failure")
	}
	if !strings.Contains(err.Error(), "readSource column Widget.CreatedById is missing from _System_RecordType") {
		t.Fatalf("validateReadSourceLegacySurfaceCompatibility() error = %v, want backing-column detail", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestValidateReadSourceLegacySurfaceCompatibilityRejectsNonCanonicalRecordTypeSourceTable(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	defs := []domainschema.TableDefinition{{
		TableName: "RecordType",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "RecordTypeShadow",
		},
	}}

	mock.ExpectQuery("SELECT TABLE_NAME FROM INFORMATION_SCHEMA\\.TABLES").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME"}).AddRow("RecordTypeShadow"))
	mock.ExpectQuery("SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLLATION_NAME FROM INFORMATION_SCHEMA\\.COLUMNS").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME", "COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE", "COLUMN_DEFAULT", "COLLATION_NAME"}))

	err = validateReadSourceLegacySurfaceCompatibility(context.Background(), db, "crm", defs, nil)
	if err == nil {
		t.Fatal("validateReadSourceLegacySurfaceCompatibility() = nil, want canonical table failure")
	}
	if !strings.Contains(err.Error(), constants.TableRecordType) {
		t.Fatalf("validateReadSourceLegacySurfaceCompatibility() error = %v, want canonical table detail", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestValidateReadSourceLegacySurfaceCompatibilityAllowsPlatformCoreRecordTypePhysicalSource(t *testing.T) {
	specs, err := parseStandardObjectsSpec()
	if err != nil {
		t.Fatalf("parseStandardObjectsSpec() error = %v", err)
	}
	recordTypeSpec := findObjectSpec(t, specs, constants.ObjectRecordType)
	sm := metadata.NewSchemaManager(persistence.NewSchemaRepository(nil, nil))
	resolver := newObjectSpecFieldResolver([]ObjectSpec{recordTypeSpec})
	defs, err := convertSpecToTableDefinitionsWithResolver([]ObjectSpec{recordTypeSpec}, sm, resolver)
	if err != nil {
		t.Fatalf("convertSpecToTableDefinitionsWithResolver() error = %v", err)
	}
	fields, err := buildFieldMetadataWithResolver("platform-core", []ObjectSpec{recordTypeSpec}, sm, false, resolver, packFieldAuthoringCompat{})
	if err != nil {
		t.Fatalf("buildFieldMetadataWithResolver() error = %v", err)
	}

	for _, unsupported := range []string{
		constants.FieldOwnerID,
		constants.FieldCreatedByID,
		constants.FieldLastModifiedByID,
		constants.FieldSystemModstamp,
	} {
		if hasColumnDefinition(defs[0].Columns, unsupported) {
			t.Fatalf("RecordType readSource def unexpectedly includes unsupported system column %s", unsupported)
		}
		if hasFieldMetadata(fields, constants.ObjectRecordType, unsupported) {
			t.Fatalf("RecordType readSource metadata unexpectedly publishes unsupported system field %s", unsupported)
		}
	}

	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	mock.ExpectQuery("SELECT TABLE_NAME FROM INFORMATION_SCHEMA\\.TABLES").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME"}).AddRow(constants.TableRecordType))
	mock.ExpectQuery("SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLLATION_NAME FROM INFORMATION_SCHEMA\\.COLUMNS").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME", "COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE", "COLUMN_DEFAULT", "COLLATION_NAME"}).
			AddRow(constants.TableRecordType, constants.FieldID, "varchar(255)", "NO", nil, nil).
			AddRow(constants.TableRecordType, constants.FieldName, "varchar(255)", "NO", nil, nil).
			AddRow(constants.TableRecordType, constants.FieldSysRecordType_DeveloperName, "varchar(255)", "YES", nil, nil).
			AddRow(constants.TableRecordType, "NamespacePrefix", "varchar(255)", "YES", nil, nil).
			AddRow(constants.TableRecordType, constants.FieldDescription, "text", "YES", nil, nil).
			AddRow(constants.TableRecordType, constants.FieldSysRecordType_BusinessProcessID, "varchar(255)", "YES", nil, nil).
			AddRow(constants.TableRecordType, "SobjectType", "varchar(255)", "NO", "", nil).
			AddRow(constants.TableRecordType, constants.FieldSysRecordType_IsActive, "tinyint(1)", "NO", "1", nil).
			AddRow(constants.TableRecordType, constants.FieldIsDeleted, "tinyint(1)", "NO", "0", nil).
			AddRow(constants.TableRecordType, constants.FieldCreatedDate, "datetime", "NO", "CURRENT_TIMESTAMP", nil).
			AddRow(constants.TableRecordType, constants.FieldLastModifiedDate, "datetime", "NO", "CURRENT_TIMESTAMP", nil))
	mock.ExpectQuery("(?s)SELECT f\\.ApiName,\\s*COALESCE\\(f\\.OwnerKind, ''\\),\\s*COALESCE\\(f\\.OwnerPack, ''\\),\\s*COALESCE\\(f\\.FormulaMode, ''\\),\\s*COALESCE\\(f\\.FormulaStorageMode, ''\\),\\s*COALESCE\\(f\\.Formula, ''\\),\\s*COALESCE\\(f\\.ProjectionConfig, ''\\)\\s*FROM _System_Field f\\s*JOIN _System_Object o ON o\\.Id = f\\.ObjectId\\s*WHERE o\\.ApiName = \\? AND COALESCE\\(f\\.IsDeleted, 0\\) = 0 AND COALESCE\\(o\\.IsDeleted, 0\\) = 0").
		WithArgs(constants.ObjectRecordType).
		WillReturnRows(sqlmock.NewRows([]string{"ApiName", "OwnerKind", "OwnerPack", "FormulaMode", "FormulaStorageMode", "Formula", "ProjectionConfig"}))

	if err := validateReadSourceLegacySurfaceCompatibility(context.Background(), db, "platform-core", defs, fields); err != nil {
		t.Fatalf("validateReadSourceLegacySurfaceCompatibility() = %v, want nil", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestValidateReadSourceLegacySurfaceCompatibilityFailsOnAuthoredActiveComputedFieldBeforePublish(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	defs := []domainschema.TableDefinition{{
		TableName: "RecordType",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
		Columns: []domainschema.ColumnDefinition{{
			Name:        "FormulaName__c",
			FormulaMode: "active",
			Formula:     "Name",
		}},
	}}
	fields := []metadata.FieldWithContext{
		{TableName: "RecordType", Field: &models.FieldMetadata{APIName: "FormulaName__c"}},
	}

	mock.ExpectQuery("SELECT TABLE_NAME FROM INFORMATION_SCHEMA\\.TABLES").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME"}).AddRow("_System_RecordType"))
	mock.ExpectQuery("SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLLATION_NAME FROM INFORMATION_SCHEMA\\.COLUMNS").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME", "COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE", "COLUMN_DEFAULT", "COLLATION_NAME"}))

	err = validateReadSourceLegacySurfaceCompatibility(context.Background(), db, "salesforce-crm", defs, fields)
	if err == nil {
		t.Fatal("validateReadSourceLegacySurfaceCompatibility() = nil, want authored active computed field failure")
	}
	if !strings.Contains(err.Error(), "declares active computed field FormulaName__c") {
		t.Fatalf("validateReadSourceLegacySurfaceCompatibility() error = %v, want authored active computed detail", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestReconcileReadSourceLegacySurfacesDropsWarmUpgradeArtifacts(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	defs := []domainschema.TableDefinition{{
		TableName: "RecordType",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
	}}
	fields := []metadata.FieldWithContext{
		{TableName: "RecordType", Field: &models.FieldMetadata{APIName: constants.FieldID}},
		{TableName: "RecordType", Field: &models.FieldMetadata{APIName: "DeveloperName"}},
	}

	mock.ExpectQuery("SELECT TABLE_NAME FROM INFORMATION_SCHEMA\\.TABLES").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME"}).AddRow("_System_RecordType"))
	mock.ExpectQuery("SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLLATION_NAME FROM INFORMATION_SCHEMA\\.COLUMNS").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME", "COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE", "COLUMN_DEFAULT", "COLLATION_NAME"}))
	mock.ExpectQuery("(?s)SELECT f\\.ApiName,\\s*COALESCE\\(f\\.OwnerKind, ''\\),\\s*COALESCE\\(f\\.OwnerPack, ''\\),\\s*COALESCE\\(f\\.FormulaMode, ''\\),\\s*COALESCE\\(f\\.FormulaStorageMode, ''\\),\\s*COALESCE\\(f\\.Formula, ''\\),\\s*COALESCE\\(f\\.ProjectionConfig, ''\\)\\s*FROM _System_Field f\\s*JOIN _System_Object o ON o\\.Id = f\\.ObjectId\\s*WHERE o\\.ApiName = \\? AND COALESCE\\(f\\.IsDeleted, 0\\) = 0 AND COALESCE\\(o\\.IsDeleted, 0\\) = 0").
		WithArgs("RecordType").
		WillReturnRows(sqlmock.NewRows([]string{"ApiName", "OwnerKind", "OwnerPack", "FormulaMode", "FormulaStorageMode", "Formula", "ProjectionConfig"}).
			AddRow("DeveloperName", constants.FieldOwnerKindPack, "salesforce-crm", "", "", "", "").
			AddRow(constants.FieldCreatedByID, constants.FieldOwnerKindPack, "salesforce-crm", "", "", "", ""))
	mock.ExpectQuery("SELECT `ApiName`, `TableType`, `KeyPrefix`, `DefaultChildRows`, `ChildCloneRules`, `ReadSource`, COALESCE\\(`ChildCloneRulesPackName`, ''\\), COALESCE\\(`ChildCloneRulesPackBusinessKey`, ''\\), COALESCE\\(`ChildCloneRulesPackManageability`, ''\\), `ChildCloneRulesPackValue`, COALESCE\\(`IsCustomized`, 0\\), COALESCE\\(`DefaultChildRowsCustomized`, 0\\), COALESCE\\(`ChildCloneRulesCustomized`, 0\\), COALESCE\\(`LockRuleCustomized`, 0\\), COALESCE\\(`IsDeleted`, 0\\), `LockRule`, `DisplayNameField` FROM `_System_Object`").
		WillReturnRows(sqlmock.NewRows([]string{"ApiName", "TableType", "KeyPrefix", "DefaultChildRows", "ChildCloneRules", "ReadSource", "ChildCloneRulesPackName", "ChildCloneRulesPackBusinessKey", "ChildCloneRulesPackManageability", "ChildCloneRulesPackValue", "IsCustomized", "DefaultChildRowsCustomized", "ChildCloneRulesCustomized", "LockRuleCustomized", "IsDeleted", "LockRule", "DisplayNameField"}).
			AddRow("RecordType", "standard_object", nil, nil, nil, `{"kind": "physical_table", "physicalTable": "_System_RecordType"}`, "", "", "", nil, 0, 0, 0, 0, 0, nil, nil))
	mock.ExpectQuery("(?s)SELECT f\\.ApiName,\\s*COALESCE\\(f\\.OwnerKind, ''\\),\\s*COALESCE\\(f\\.OwnerPack, ''\\),\\s*COALESCE\\(f\\.FormulaMode, ''\\),\\s*COALESCE\\(f\\.FormulaStorageMode, ''\\),\\s*COALESCE\\(f\\.Formula, ''\\),\\s*COALESCE\\(f\\.ProjectionConfig, ''\\)\\s*FROM _System_Field f\\s*JOIN _System_Object o ON o\\.Id = f\\.ObjectId\\s*WHERE o\\.ApiName = \\? AND COALESCE\\(f\\.IsDeleted, 0\\) = 0 AND COALESCE\\(o\\.IsDeleted, 0\\) = 0").
		WithArgs("RecordType").
		WillReturnRows(sqlmock.NewRows([]string{"ApiName", "OwnerKind", "OwnerPack", "FormulaMode", "FormulaStorageMode", "Formula", "ProjectionConfig"}).
			AddRow("DeveloperName", constants.FieldOwnerKindPack, "salesforce-crm", "", "", "", "").
			AddRow(constants.FieldCreatedByID, constants.FieldOwnerKindPack, "salesforce-crm", "", "", "", ""))
	mock.ExpectExec(regexp.QuoteMeta("UPDATE `_System_Table` SET `IsDeleted` = 1, `LastModifiedDate` = CURRENT_TIMESTAMP WHERE `TableName` = ? AND COALESCE(`IsDeleted`, 0) = 0")).
		WithArgs("RecordType").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(regexp.QuoteMeta(
		"UPDATE _System_Field f\n"+
			"JOIN _System_Object o ON o.Id = f.ObjectId\n"+
			"SET f.IsDeleted = 1, f.LastModifiedDate = CURRENT_TIMESTAMP\n"+
			"WHERE o.ApiName = ? AND f.ApiName = ? AND COALESCE(f.IsDeleted, 0) = 0\n"+
			"  AND ((COALESCE(f.OwnerKind, '') = ? AND COALESCE(f.OwnerPack, '') = ?)\n"+
			"    OR (COALESCE(f.OwnerKind, '') = ? AND COALESCE(f.OwnerPack, '') = ?))",
	)).
		WithArgs("RecordType", constants.FieldCreatedByID, constants.FieldOwnerKindPack, "salesforce-crm", constants.FieldOwnerKindPackRemoving, "salesforce-crm").
		WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec(regexp.QuoteMeta("DROP TABLE IF EXISTS `RecordType`")).
		WillReturnResult(sqlmock.NewResult(0, 0))

	if err := reconcileReadSourceLegacySurfaces(context.Background(), db, "salesforce-crm", defs, fields); err != nil {
		t.Fatalf("reconcileReadSourceLegacySurfaces() error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestReconcileReadSourceLegacySurfacesFailsOnUnsupportedLiveAdminField(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	defs := []domainschema.TableDefinition{{
		TableName: "RecordType",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
	}}
	fields := []metadata.FieldWithContext{
		{TableName: "RecordType", Field: &models.FieldMetadata{APIName: constants.FieldID}},
	}

	mock.ExpectQuery("SELECT TABLE_NAME FROM INFORMATION_SCHEMA\\.TABLES").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME"}).AddRow("_System_RecordType"))
	mock.ExpectQuery("SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLLATION_NAME FROM INFORMATION_SCHEMA\\.COLUMNS").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME", "COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE", "COLUMN_DEFAULT", "COLLATION_NAME"}))
	mock.ExpectQuery("(?s)SELECT f\\.ApiName,\\s*COALESCE\\(f\\.OwnerKind, ''\\),\\s*COALESCE\\(f\\.OwnerPack, ''\\),\\s*COALESCE\\(f\\.FormulaMode, ''\\),\\s*COALESCE\\(f\\.FormulaStorageMode, ''\\),\\s*COALESCE\\(f\\.Formula, ''\\),\\s*COALESCE\\(f\\.ProjectionConfig, ''\\)\\s*FROM _System_Field f\\s*JOIN _System_Object o ON o\\.Id = f\\.ObjectId\\s*WHERE o\\.ApiName = \\? AND COALESCE\\(f\\.IsDeleted, 0\\) = 0 AND COALESCE\\(o\\.IsDeleted, 0\\) = 0").
		WithArgs("RecordType").
		WillReturnRows(sqlmock.NewRows([]string{"ApiName", "OwnerKind", "OwnerPack", "FormulaMode", "FormulaStorageMode", "Formula", "ProjectionConfig"}).
			AddRow("CustomScore__c", constants.FieldOwnerKindAdmin, "", "", "", "", ""))

	err = reconcileReadSourceLegacySurfaces(context.Background(), db, "salesforce-crm", defs, fields)
	if err == nil {
		t.Fatal("reconcileReadSourceLegacySurfaces() error = nil, want unsupported admin field failure")
	}
	if !strings.Contains(err.Error(), "unsupported live fields") || !strings.Contains(err.Error(), "CustomScore__c") {
		t.Fatalf("error = %v, want unsupported admin field detail", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestReconcileReadSourceLegacySurfacesFailsOnActiveComputedExpectedField(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	defs := []domainschema.TableDefinition{{
		TableName: "RecordType",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
	}}
	fields := []metadata.FieldWithContext{
		{TableName: "RecordType", Field: &models.FieldMetadata{APIName: "FormulaName__c"}},
	}

	mock.ExpectQuery("SELECT TABLE_NAME FROM INFORMATION_SCHEMA\\.TABLES").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME"}).AddRow("_System_RecordType"))
	mock.ExpectQuery("SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLLATION_NAME FROM INFORMATION_SCHEMA\\.COLUMNS").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME", "COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE", "COLUMN_DEFAULT", "COLLATION_NAME"}))
	mock.ExpectQuery("(?s)SELECT f\\.ApiName,\\s*COALESCE\\(f\\.OwnerKind, ''\\),\\s*COALESCE\\(f\\.OwnerPack, ''\\),\\s*COALESCE\\(f\\.FormulaMode, ''\\),\\s*COALESCE\\(f\\.FormulaStorageMode, ''\\),\\s*COALESCE\\(f\\.Formula, ''\\),\\s*COALESCE\\(f\\.ProjectionConfig, ''\\)\\s*FROM _System_Field f\\s*JOIN _System_Object o ON o\\.Id = f\\.ObjectId\\s*WHERE o\\.ApiName = \\? AND COALESCE\\(f\\.IsDeleted, 0\\) = 0 AND COALESCE\\(o\\.IsDeleted, 0\\) = 0").
		WithArgs("RecordType").
		WillReturnRows(sqlmock.NewRows([]string{"ApiName", "OwnerKind", "OwnerPack", "FormulaMode", "FormulaStorageMode", "Formula", "ProjectionConfig"}).
			AddRow("FormulaName__c", constants.FieldOwnerKindPack, "salesforce-crm", "active", "", "Name", ""))

	err = reconcileReadSourceLegacySurfaces(context.Background(), db, "salesforce-crm", defs, fields)
	if err == nil {
		t.Fatal("reconcileReadSourceLegacySurfaces() error = nil, want active computed field failure")
	}
	if !strings.Contains(err.Error(), "active computed metadata") || !strings.Contains(err.Error(), "FormulaName__c") {
		t.Fatalf("error = %v, want active computed field detail", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestReconcileReadSourceLegacySurfacesFailsWhenLiveReadSourceDidNotSwitch(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	defs := []domainschema.TableDefinition{{
		TableName: "RecordType",
		ReadSource: &models.ObjectReadSourceConfig{
			Kind:          "physical_table",
			PhysicalTable: "_System_RecordType",
		},
	}}
	fields := []metadata.FieldWithContext{
		{TableName: "RecordType", Field: &models.FieldMetadata{APIName: constants.FieldID}},
		{TableName: "RecordType", Field: &models.FieldMetadata{APIName: "DeveloperName"}},
	}

	mock.ExpectQuery("SELECT TABLE_NAME FROM INFORMATION_SCHEMA\\.TABLES").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME"}).AddRow("_System_RecordType"))
	mock.ExpectQuery("SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLLATION_NAME FROM INFORMATION_SCHEMA\\.COLUMNS").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME", "COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE", "COLUMN_DEFAULT", "COLLATION_NAME"}))
	mock.ExpectQuery("(?s)SELECT f\\.ApiName,\\s*COALESCE\\(f\\.OwnerKind, ''\\),\\s*COALESCE\\(f\\.OwnerPack, ''\\),\\s*COALESCE\\(f\\.FormulaMode, ''\\),\\s*COALESCE\\(f\\.FormulaStorageMode, ''\\),\\s*COALESCE\\(f\\.Formula, ''\\),\\s*COALESCE\\(f\\.ProjectionConfig, ''\\)\\s*FROM _System_Field f\\s*JOIN _System_Object o ON o\\.Id = f\\.ObjectId\\s*WHERE o\\.ApiName = \\? AND COALESCE\\(f\\.IsDeleted, 0\\) = 0 AND COALESCE\\(o\\.IsDeleted, 0\\) = 0").
		WithArgs("RecordType").
		WillReturnRows(sqlmock.NewRows([]string{"ApiName", "OwnerKind", "OwnerPack", "FormulaMode", "FormulaStorageMode", "Formula", "ProjectionConfig"}).
			AddRow("DeveloperName", constants.FieldOwnerKindPack, "salesforce-crm", "", "", "", ""))
	mock.ExpectQuery("SELECT `ApiName`, `TableType`, `KeyPrefix`, `DefaultChildRows`, `ChildCloneRules`, `ReadSource`, COALESCE\\(`ChildCloneRulesPackName`, ''\\), COALESCE\\(`ChildCloneRulesPackBusinessKey`, ''\\), COALESCE\\(`ChildCloneRulesPackManageability`, ''\\), `ChildCloneRulesPackValue`, COALESCE\\(`IsCustomized`, 0\\), COALESCE\\(`DefaultChildRowsCustomized`, 0\\), COALESCE\\(`ChildCloneRulesCustomized`, 0\\), COALESCE\\(`LockRuleCustomized`, 0\\), COALESCE\\(`IsDeleted`, 0\\), `LockRule`, `DisplayNameField` FROM `_System_Object`").
		WillReturnRows(sqlmock.NewRows([]string{"ApiName", "TableType", "KeyPrefix", "DefaultChildRows", "ChildCloneRules", "ReadSource", "ChildCloneRulesPackName", "ChildCloneRulesPackBusinessKey", "ChildCloneRulesPackManageability", "ChildCloneRulesPackValue", "IsCustomized", "DefaultChildRowsCustomized", "ChildCloneRulesCustomized", "LockRuleCustomized", "IsDeleted", "LockRule", "DisplayNameField"}).
			AddRow("RecordType", "standard_object", nil, nil, nil, nil, "", "", "", nil, 1, 0, 0, 0, 0, nil, nil))

	err = reconcileReadSourceLegacySurfaces(context.Background(), db, "salesforce-crm", defs, fields)
	if err == nil {
		t.Fatal("reconcileReadSourceLegacySurfaces() error = nil, want live ReadSource mismatch failure")
	}
	if !strings.Contains(err.Error(), "ReadSource") || !strings.Contains(err.Error(), "before dropping legacy surface") {
		t.Fatalf("error = %v, want live ReadSource mismatch detail", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func hasFieldMetadata(fields []metadata.FieldWithContext, objectAPIName string, fieldAPIName string) bool {
	for _, field := range fields {
		if !strings.EqualFold(field.TableName, objectAPIName) || field.Field == nil {
			continue
		}
		if strings.EqualFold(field.Field.APIName, fieldAPIName) {
			return true
		}
	}
	return false
}