package packs

import (
	"context"
	"database/sql/driver"
	"os"
	"path/filepath"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	domainschema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/db/pkg/schemashape"
)

func TestVerifyPackObjectPhysicalSchemaAllowsDeclaredPeerPackField(t *testing.T) {
	sourceDir := writeCrossPackFieldTestSource(t, "peer-pack", "PeerField")

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

	expectVerifyPackObjectPhysicalSchemaBaseQueries(mock, []driver.Value{
		"Issue", "PeerField", "pack", "peer-pack",
	}, []driver.Value{
		"Issue", "PeerField", "varchar(255)", "YES", nil, nil,
	})
	expectInstalledPackStatusLookup(mock, "peer-pack", []driver.Value{"active", 0})
	expectInstalledPackDeclarationLookup(mock, "peer-pack", installedPackDeclarationValues(t, "peer-pack", sourceDir, "active"))

	err = verifyPackObjectPhysicalSchema(db, "crm-pack", []domainschema.TableDefinition{{
		TableName: "Issue",
		TableType: "standard_object",
		Category:  "jira",
	}}, ManageabilityLocked)
	if err != nil {
		t.Fatalf("verifyPackObjectPhysicalSchema: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyPackObjectPhysicalSchemaAllowsDeclaredPeerFieldWhenActivePackNeedsLiveConnectorAssociationSchema(t *testing.T) {
	sourceDir := writeCrossPackConnectorAssociationLookupTestSource(t)
	row := installedPackDeclarationValuesForConnectorAssociationLookup(t, sourceDir)

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

	expectVerifyPackObjectPhysicalSchemaBaseQueries(mock, []driver.Value{
		"Issue", "PeerField__c", "pack", "jira-like-sample",
	}, []driver.Value{
		"Issue", "PeerField__c", "varchar(255)", "YES", nil, nil,
	})
	expectInstalledPackStatusLookup(mock, "jira-like-sample", []driver.Value{"active", 0})
	expectInstalledPackDeclarationLookup(mock, "jira-like-sample", row)
	expectConnectorAssociationLookupSchemaLoad(mock)

	err = verifyPackObjectPhysicalSchema(db, "crm-pack", []domainschema.TableDefinition{{
		TableName: "Issue",
		TableType: "standard_object",
		Category:  "jira",
	}}, ManageabilityLocked)
	if err != nil {
		t.Fatalf("verifyPackObjectPhysicalSchema: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyPackObjectPhysicalSchemaRejectsDeclaredPeerPackFieldWithConvergedStaleHashes(t *testing.T) {
	sourceDir := writeCrossPackFieldTestSource(t, "peer-pack", "PeerField__c")

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

	expectVerifyPackObjectPhysicalSchemaBaseQueries(mock, []driver.Value{
		"Issue", "PeerField__c", "pack", "peer-pack",
	}, []driver.Value{
		"Issue", "PeerField__c", "varchar(255)", "YES", nil, nil,
	})
	expectInstalledPackStatusLookup(mock, "peer-pack", []driver.Value{"active", 0})
	mock.ExpectQuery("SELECT Name, Category, Status, COALESCE\\(IsDeleted, 0\\), ManifestFormatVersion, SourceKind, SourceURI, SourceRevision, AppliedPayloadHash, DesiredPayloadHash FROM _System_InstalledPack").
		WithArgs("peer-pack").
		WillReturnRows(sqlmock.NewRows([]string{"Name", "Category", "Status", "IsDeleted", "ManifestFormatVersion", "SourceKind", "SourceURI", "SourceRevision", "AppliedPayloadHash", "DesiredPayloadHash"}).
			AddRow("peer-pack", "app-pack", "active", 0, 1, "local_path", sourceDir, "", "stale-hash", "stale-hash"))

	err = verifyPackObjectPhysicalSchema(db, "crm-pack", []domainschema.TableDefinition{{
		TableName: "Issue",
		TableType: "standard_object",
		Category:  "jira",
	}}, ManageabilityLocked)
	if err == nil {
		t.Fatal("verifyPackObjectPhysicalSchema = nil, want stale-hash peer-pack rejection")
	}
	if !strings.Contains(err.Error(), "AppliedPayloadHash") {
		t.Fatalf("verifyPackObjectPhysicalSchema error = %v, want AppliedPayloadHash drift detail", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyPackObjectPhysicalSchemaAllowsDeclaredPeerPackFieldWithinReconcileOverride(t *testing.T) {
	sourceDir := writeCrossPackFieldTestSource(t, "peer-pack", "PeerField__c")
	peerPack, err := ResolveInstalledSourceBackedPack("peer-pack", AppPack, "local_path", sourceDir)
	if err != nil {
		t.Fatalf("resolve installed pack: %v", err)
	}

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

	expectVerifyPackObjectPhysicalSchemaBaseQueries(mock, []driver.Value{
		"Issue", "PeerField__c", "pack", "peer-pack",
	}, []driver.Value{
		"Issue", "PeerField__c", "varchar(255)", "YES", nil, nil,
	})
	expectInstalledPackStatusLookup(mock, "peer-pack", []driver.Value{"active", 0})

	ctx := WithPeerDeclarationOverrides(context.Background(), map[string]*SourceBackedPack{
		"peer-pack": peerPack,
	})
	err = verifyPackObjectPhysicalSchemaWithContext(ctx, db, "crm-pack", []domainschema.TableDefinition{{
		TableName: "Issue",
		TableType: "standard_object",
		Category:  "jira",
	}}, ManageabilityLocked)
	if err != nil {
		t.Fatalf("verifyPackObjectPhysicalSchemaWithContext declared peer-pack field: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyPackObjectPhysicalSchemaRejectsUndeclaredPeerPackField(t *testing.T) {
	sourceDir := writeCrossPackFieldTestSource(t, "peer-pack", "PeerField__c")

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

	expectVerifyPackObjectPhysicalSchemaBaseQueries(mock, []driver.Value{
		"Issue", "LegacyField__c", "pack", "peer-pack",
	}, []driver.Value{
		"Issue", "LegacyField__c", "varchar(255)", "YES", nil, nil,
	})
	expectInstalledPackStatusLookup(mock, "peer-pack", []driver.Value{"active", 0})
	expectInstalledPackDeclarationLookup(mock, "peer-pack", installedPackDeclarationValues(t, "peer-pack", sourceDir, "active"))

	err = verifyPackObjectPhysicalSchema(db, "crm-pack", []domainschema.TableDefinition{{
		TableName: "Issue",
		TableType: "standard_object",
		Category:  "jira",
	}}, ManageabilityLocked)
	if err == nil {
		t.Fatal("verifyPackObjectPhysicalSchema = nil, want undeclared peer-pack drift")
	}
	if !strings.Contains(err.Error(), "Issue.LegacyField__c") {
		t.Fatalf("verifyPackObjectPhysicalSchema error = %v, want field drift for Issue.LegacyField__c", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

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

	expectVerifyPackObjectPhysicalSchemaBaseQueries(mock, []driver.Value{
		"Issue", "LegacyField__c", constants.FieldOwnerKindPackRemoving, "peer-pack",
	}, nil)

	err = verifyPackObjectPhysicalSchema(db, "crm-pack", []domainschema.TableDefinition{{
		TableName: "Issue",
		TableType: "standard_object",
		Category:  "jira",
	}}, ManageabilityLocked)
	if err == nil {
		t.Fatal("verifyPackObjectPhysicalSchema = nil, want pack_removing drift when column is already dropped")
	}
	if !strings.Contains(err.Error(), "Issue.LegacyField__c") {
		t.Fatalf("verifyPackObjectPhysicalSchema error = %v, want field drift for Issue.LegacyField__c", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

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

	expectVerifyPackObjectPhysicalSchemaBaseQueries(mock, []driver.Value{
		"Issue", "LegacyField__c", constants.FieldOwnerKindPackRemoving, "peer-pack",
	}, []driver.Value{
		"Issue", "LegacyField__c", "varchar(255)", "YES", nil, nil,
	})

	err = verifyPackObjectPhysicalSchema(db, "crm-pack", []domainschema.TableDefinition{{
		TableName: "Issue",
		TableType: "standard_object",
		Category:  "jira",
	}}, ManageabilityLocked)
	if err == nil {
		t.Fatal("verifyPackObjectPhysicalSchema = nil, want pack_removing drift while physical column still exists")
	}
	if !strings.Contains(err.Error(), "Issue.LegacyField__c") {
		t.Fatalf("verifyPackObjectPhysicalSchema error = %v, want field drift for Issue.LegacyField__c", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

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

	expectVerifyPackObjectPhysicalSchemaBaseQueries(mock, []driver.Value{
		"Issue", "LegacyField__c", constants.FieldOwnerKindPackRemoving, "crm-pack",
	}, nil)

	err = verifyPackObjectPhysicalSchema(db, "crm-pack", []domainschema.TableDefinition{{
		TableName: "Issue",
		TableType: "standard_object",
		Category:  "jira",
	}}, ManageabilityLocked)
	if err == nil {
		t.Fatal("verifyPackObjectPhysicalSchema = nil, want same-pack pack_removing drift")
	}
	if !strings.Contains(err.Error(), "Issue.LegacyField__c") {
		t.Fatalf("verifyPackObjectPhysicalSchema error = %v, want field drift for Issue.LegacyField__c", err)
	}
	if !strings.Contains(err.Error(), "extra live pack-owned field") {
		t.Fatalf("verifyPackObjectPhysicalSchema error = %v, want same-pack fail-closed drift", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

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

	expectVerifyPackObjectPhysicalSchemaBaseQueries(mock, []driver.Value{
		"Issue", "LegacyField__c", constants.FieldOwnerKindPackRemoving, "CRM-Pack",
	}, nil)

	err = verifyPackObjectPhysicalSchema(db, "crm-pack", []domainschema.TableDefinition{{
		TableName: "Issue",
		TableType: "standard_object",
		Category:  "jira",
	}}, ManageabilityLocked)
	if err == nil {
		t.Fatal("verifyPackObjectPhysicalSchema = nil, want same-pack pack_removing drift with OwnerPack case drift")
	}
	if !strings.Contains(err.Error(), "Issue.LegacyField__c") {
		t.Fatalf("verifyPackObjectPhysicalSchema error = %v, want field drift for Issue.LegacyField__c", err)
	}
	if !strings.Contains(err.Error(), "extra live pack-owned field") {
		t.Fatalf("verifyPackObjectPhysicalSchema error = %v, want same-pack fail-closed drift", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyOwnerlessObjectSpecPartialCompatibilityAllowsDeclaredPeerPackField(t *testing.T) {
	sourceDir := writeCrossPackFieldTestSource(t, "peer-pack", "PeerField__c")

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

	expectInstalledPackStatusLookup(mock, "peer-pack", []driver.Value{"active", 0})
	expectInstalledPackDeclarationLookup(mock, "peer-pack", installedPackDeclarationValues(t, "peer-pack", sourceDir, "active"))

	err = verifyOwnerlessObjectSpecPartialCompatibility(
		context.Background(),
		db,
		"crm-pack",
		crossPackRetryCompatibilityDefinition(),
		map[string]bool{"issue": true},
		nil,
		map[string]packTableRegistryRow{"issue": {tableType: "standard_object", category: "jira", managed: true}},
		map[string]packObjectRegistryRow{"issue": {tableType: "standard_object"}},
		map[string]map[string]packFieldRegistryRow{
			"issue": {
				"id":           {fieldAPIName: "Id", ownerKind: "pack", ownerPack: "crm-pack"},
				"subject":      {fieldAPIName: "Subject", ownerKind: "pack", ownerPack: "crm-pack"},
				"peerfield__c": {fieldAPIName: "PeerField__c", ownerKind: "pack", ownerPack: "peer-pack"},
			},
		},
	)
	if err != nil {
		t.Fatalf("verifyOwnerlessObjectSpecPartialCompatibility declared peer-pack field: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyOwnerlessObjectSpecPartialCompatibilityRejectsUndeclaredPeerPackField(t *testing.T) {
	sourceDir := writeCrossPackFieldTestSource(t, "peer-pack", "PeerField__c")

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

	expectInstalledPackStatusLookup(mock, "peer-pack", []driver.Value{"active", 0})
	expectInstalledPackDeclarationLookup(mock, "peer-pack", installedPackDeclarationValues(t, "peer-pack", sourceDir, "active"))

	err = verifyOwnerlessObjectSpecPartialCompatibility(
		context.Background(),
		db,
		"crm-pack",
		crossPackRetryCompatibilityDefinition(),
		map[string]bool{"issue": true},
		nil,
		map[string]packTableRegistryRow{"issue": {tableType: "standard_object", category: "jira", managed: true}},
		map[string]packObjectRegistryRow{"issue": {tableType: "standard_object"}},
		map[string]map[string]packFieldRegistryRow{
			"issue": {
				"id":             {fieldAPIName: "Id", ownerKind: "pack", ownerPack: "crm-pack"},
				"subject":        {fieldAPIName: "Subject", ownerKind: "pack", ownerPack: "crm-pack"},
				"legacyfield__c": {fieldAPIName: "LegacyField__c", ownerKind: "pack", ownerPack: "peer-pack"},
			},
		},
	)
	if err == nil {
		t.Fatal("verifyOwnerlessObjectSpecPartialCompatibility = nil, want undeclared peer-pack drift")
	}
	if !strings.Contains(err.Error(), "LegacyField__c") {
		t.Fatalf("verifyOwnerlessObjectSpecPartialCompatibility error = %v, want field drift for Issue.LegacyField__c", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyOwnerlessObjectSpecPartialCompatibilityRejectsDeclaredPeerPackFieldWithConvergedStaleHashes(t *testing.T) {
	sourceDir := writeCrossPackFieldTestSource(t, "peer-pack", "PeerField__c")

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

	expectInstalledPackStatusLookup(mock, "peer-pack", []driver.Value{"active", 0})
	mock.ExpectQuery("SELECT Name, Category, Status, COALESCE\\(IsDeleted, 0\\), ManifestFormatVersion, SourceKind, SourceURI, SourceRevision, AppliedPayloadHash, DesiredPayloadHash FROM _System_InstalledPack").
		WithArgs("peer-pack").
		WillReturnRows(sqlmock.NewRows([]string{"Name", "Category", "Status", "IsDeleted", "ManifestFormatVersion", "SourceKind", "SourceURI", "SourceRevision", "AppliedPayloadHash", "DesiredPayloadHash"}).
			AddRow("peer-pack", "app-pack", "active", 0, 1, "local_path", sourceDir, "", "stale-hash", "stale-hash"))

	err = verifyOwnerlessObjectSpecPartialCompatibility(
		context.Background(),
		db,
		"crm-pack",
		crossPackRetryCompatibilityDefinition(),
		map[string]bool{"issue": true},
		nil,
		map[string]packTableRegistryRow{"issue": {tableType: "standard_object", category: "jira", managed: true}},
		map[string]packObjectRegistryRow{"issue": {tableType: "standard_object"}},
		map[string]map[string]packFieldRegistryRow{
			"issue": {
				"id":           {fieldAPIName: "Id", ownerKind: "pack", ownerPack: "crm-pack"},
				"subject":      {fieldAPIName: "Subject", ownerKind: "pack", ownerPack: "crm-pack"},
				"peerfield__c": {fieldAPIName: "PeerField__c", ownerKind: "pack", ownerPack: "peer-pack"},
			},
		},
	)
	if err == nil {
		t.Fatal("verifyOwnerlessObjectSpecPartialCompatibility = nil, want stale-hash peer-pack rejection")
	}
	if !strings.Contains(err.Error(), "AppliedPayloadHash") {
		t.Fatalf("verifyOwnerlessObjectSpecPartialCompatibility error = %v, want AppliedPayloadHash drift detail", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyOwnerlessObjectSpecPartialCompatibilityAllowsDeclaredPeerPackFieldWithinReconcileOverride(t *testing.T) {
	sourceDir := writeCrossPackFieldTestSource(t, "peer-pack", "PeerField__c")
	peerPack, err := ResolveInstalledSourceBackedPack("peer-pack", AppPack, "local_path", sourceDir)
	if err != nil {
		t.Fatalf("resolve installed pack: %v", err)
	}

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

	expectInstalledPackStatusLookup(mock, "peer-pack", []driver.Value{"active", 0})

	ctx := WithPeerDeclarationOverrides(context.Background(), map[string]*SourceBackedPack{
		"peer-pack": peerPack,
	})
	err = verifyOwnerlessObjectSpecPartialCompatibility(
		ctx,
		db,
		"crm-pack",
		crossPackRetryCompatibilityDefinition(),
		map[string]bool{"issue": true},
		nil,
		map[string]packTableRegistryRow{"issue": {tableType: "standard_object", category: "jira", managed: true}},
		map[string]packObjectRegistryRow{"issue": {tableType: "standard_object"}},
		map[string]map[string]packFieldRegistryRow{
			"issue": {
				"id":           {fieldAPIName: "Id", ownerKind: "pack", ownerPack: "crm-pack"},
				"subject":      {fieldAPIName: "Subject", ownerKind: "pack", ownerPack: "crm-pack"},
				"peerfield__c": {fieldAPIName: "PeerField__c", ownerKind: "pack", ownerPack: "peer-pack"},
			},
		},
	)
	if err != nil {
		t.Fatalf("verifyOwnerlessObjectSpecPartialCompatibility declared peer-pack field: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

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

	err = verifyOwnerlessObjectSpecPartialCompatibility(
		context.Background(),
		db,
		"crm-pack",
		crossPackRetryCompatibilityDefinition(),
		map[string]bool{"issue": true},
		map[string]map[string]schemashape.PhysicalColumnShape{"issue": {}},
		map[string]packTableRegistryRow{"issue": {tableType: "standard_object", category: "jira", managed: true}},
		map[string]packObjectRegistryRow{"issue": {tableType: "standard_object"}},
		map[string]map[string]packFieldRegistryRow{
			"issue": {
				"id":             {fieldAPIName: "Id", ownerKind: "pack", ownerPack: "crm-pack"},
				"subject":        {fieldAPIName: "Subject", ownerKind: "pack", ownerPack: "crm-pack"},
				"legacyfield__c": {fieldAPIName: "LegacyField__c", ownerKind: constants.FieldOwnerKindPackRemoving, ownerPack: "peer-pack"},
			},
		},
	)
	if err == nil {
		t.Fatal("verifyOwnerlessObjectSpecPartialCompatibility = nil, want pack_removing drift after phase-1 drop")
	}
	if !strings.Contains(err.Error(), "LegacyField__c") {
		t.Fatalf("verifyOwnerlessObjectSpecPartialCompatibility error = %v, want field drift for Issue.LegacyField__c", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

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

	err = verifyOwnerlessObjectSpecPartialCompatibility(
		context.Background(),
		db,
		"crm-pack",
		crossPackRetryCompatibilityDefinition(),
		map[string]bool{"issue": true},
		map[string]map[string]schemashape.PhysicalColumnShape{
			"issue": {
				"legacyfield__c": {ColumnType: "varchar(255)", Nullable: true},
			},
		},
		map[string]packTableRegistryRow{"issue": {tableType: "standard_object", category: "jira", managed: true}},
		map[string]packObjectRegistryRow{"issue": {tableType: "standard_object"}},
		map[string]map[string]packFieldRegistryRow{
			"issue": {
				"id":             {fieldAPIName: "Id", ownerKind: "pack", ownerPack: "crm-pack"},
				"subject":        {fieldAPIName: "Subject", ownerKind: "pack", ownerPack: "crm-pack"},
				"legacyfield__c": {fieldAPIName: "LegacyField__c", ownerKind: constants.FieldOwnerKindPackRemoving, ownerPack: "peer-pack"},
			},
		},
	)
	if err == nil {
		t.Fatal("verifyOwnerlessObjectSpecPartialCompatibility = nil, want pack_removing drift while physical column still exists")
	}
	if !strings.Contains(err.Error(), "LegacyField__c") {
		t.Fatalf("verifyOwnerlessObjectSpecPartialCompatibility error = %v, want field drift for Issue.LegacyField__c", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func crossPackRetryCompatibilityDefinition() domainschema.TableDefinition {
	return domainschema.TableDefinition{
		TableName: "Issue",
		TableType: "standard_object",
		Category:  "jira",
		Columns: []domainschema.ColumnDefinition{
			{Name: "Id", Type: "VARCHAR(255)", PrimaryKey: true},
			{Name: "Subject", Type: "VARCHAR(255)"},
		},
	}
}

func expectVerifyPackObjectPhysicalSchemaBaseQueries(mock sqlmock.Sqlmock, fieldRow []driver.Value, physicalColumnRow []driver.Value) {
	mock.ExpectQuery("SELECT TABLE_NAME FROM INFORMATION_SCHEMA\\.TABLES").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME"}).AddRow("Issue"))

	columnRows := sqlmock.NewRows([]string{"TABLE_NAME", "COLUMN_NAME", "COLUMN_TYPE", "IS_NULLABLE", "COLUMN_DEFAULT", "COLLATION_NAME"})
	if physicalColumnRow != nil {
		columnRows.AddRow(physicalColumnRow...)
	}
	mock.ExpectQuery("SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLLATION_NAME FROM INFORMATION_SCHEMA\\.COLUMNS").
		WillReturnRows(columnRows)

	mock.ExpectQuery("SELECT TABLE_NAME, INDEX_NAME, NON_UNIQUE, GROUP_CONCAT\\(COLUMN_NAME ORDER BY SEQ_IN_INDEX\\)").
		WillReturnRows(sqlmock.NewRows([]string{"TABLE_NAME", "INDEX_NAME", "NON_UNIQUE", "cols"}))
	mock.ExpectQuery("SELECT `TableName`, `TableType`, COALESCE\\(`Category`, ''\\), COALESCE\\(`IsManaged`, 0\\), COALESCE\\(`IsDeleted`, 0\\) FROM `_System_Table`").
		WillReturnRows(sqlmock.NewRows([]string{"TableName", "TableType", "Category", "IsManaged", "IsDeleted"}).
			AddRow("Issue", "standard_object", "jira", 1, 0))
	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("Issue", "standard_object", nil, nil, nil, nil, "", "", "", nil, 0, 0, 0, 0, 0, nil, nil))
	if len(fieldRow) == 4 {
		fieldRow = append(append([]driver.Value(nil), fieldRow...), 0, nil)
	}
	mock.ExpectQuery("SELECT o.`ApiName`, f.`ApiName`, COALESCE\\(f.`OwnerKind`, ''\\), COALESCE\\(f.`OwnerPack`, ''\\), COALESCE\\(f.`IsCustomized`, 0\\), f.`MaxLength` FROM `_System_Field`").
		WillReturnRows(sqlmock.NewRows([]string{"APIName", "ApiName", "OwnerKind", "OwnerPack", "IsCustomized", "MaxLength"}).AddRow(fieldRow...))
}

func installedPackDeclarationValues(t *testing.T, packName, sourceDir, status string) []driver.Value {
	t.Helper()

	pack, err := ResolveInstalledSourceBackedPack(packName, AppPack, "local_path", sourceDir)
	if err != nil {
		t.Fatalf("resolve installed pack: %v", err)
	}
	hash, err := PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("payload hash: %v", err)
	}
	return []driver.Value{packName, "app-pack", status, 0, 1, "local_path", sourceDir, "", hash, hash}
}

func installedPackDeclarationValuesForConnectorAssociationLookup(t *testing.T, sourceDir string) []driver.Value {
	t.Helper()

	pack, err := ResolveInstalledPack(
		InstalledPackResolverInput{
			Name:                  "jira-like-sample",
			Category:              AppPack,
			ManifestFormatVersion: SupportedManifestFormatVersion,
			SourceKind:            string(PackSourceKindLocalPath),
			SourceURI:             sourceDir,
		},
		InstalledPackResolverOptions{
			AllowBundledFallback: false,
			ObjectSchemaResolver: connectorAssociationLookupSchemaResolverForTest,
		},
	)
	if err != nil {
		t.Fatalf("resolve installed pack with live connector association schema: %v", err)
	}
	hash, err := PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("payload hash: %v", err)
	}
	return []driver.Value{"jira-like-sample", "app-pack", "active", 0, SupportedManifestFormatVersion, string(PackSourceKindLocalPath), sourceDir, "", hash, hash}
}

func connectorAssociationLookupSchemaResolverForTest(apiName string) *models.ObjectMetadata {
	if strings.TrimSpace(apiName) != "Opportunity" {
		return nil
	}
	return &models.ObjectMetadata{
		APIName: "Opportunity",
		Fields: []models.FieldMetadata{{
			APIName:     "AccountId",
			Type:        constants.FieldTypeReference,
			ReferenceTo: []string{"Account"},
		}},
	}
}

func writeCrossPackFieldTestSource(t *testing.T, name string, fieldAPINames ...string) string {
	t.Helper()

	root := t.TempDir()
	objectsDir := filepath.Join(root, "objects")
	if err := os.MkdirAll(objectsDir, 0o755); err != nil {
		t.Fatalf("mkdir objects dir: %v", err)
	}

	manifest := `{
  "manifestFormatVersion": "1",
  "name": "` + name + `",
  "displayName": "Peer Pack",
  "category": "app-pack",
  "version": "0.1.0",
  "platformCompatibility": {
    "min": "0.2.0",
    "maxExclusive": "0.3.0"
  },
  "dependencies": [],
  "components": [
    {
      "kind": "object_spec",
      "businessKey": "issue",
      "manageability": "locked",
      "path": "objects/issue.json"
    }
  ]
}`
	if err := os.WriteFile(filepath.Join(root, "pack.json"), []byte(manifest), 0o644); err != nil {
		t.Fatalf("write pack manifest: %v", err)
	}

	fields := ``
	for _, fieldAPIName := range fieldAPINames {
		fields += `,
      {
        "apiName": "` + fieldAPIName + `",
        "label": "` + fieldAPIName + `",
        "type": "string"
      }`
	}
	objectSpec := `[
  {
    "apiName": "Issue",
    "label": "Issue",
    "pluralLabel": "Issues",
    "tableType": "standard_object",
    "category": "jira",
    "sharingModel": "Private",
    "enableHierarchySharing": false,
    "description": "Issue work item",
    "fields": [
      {
        "apiName": "Subject",
        "label": "Subject",
        "type": "string",
        "required": true,
        "isNameField": true
      }` + fields + `
    ]
  }
]`
	if err := os.WriteFile(filepath.Join(objectsDir, "issue.json"), []byte(objectSpec), 0o644); err != nil {
		t.Fatalf("write object spec: %v", err)
	}

	return root
}

func writeCrossPackConnectorAssociationLookupTestSource(t *testing.T) string {
	t.Helper()

	root := writePackFixture(t, "cross-pack-connector-association-lookup", validPackManifest(`[
    {
      "kind": "object_spec",
      "businessKey": "crm",
      "manageability": "locked",
      "path": "objects/crm.json"
    },
    {"kind":"standard_flow","businessKey":"Opportunity_To_Remote","manageability":"defaults_only","payload":{"start":{"object":"Opportunity"},"actionCalls":[{"name":"PushOpportunity","actionType":"PushToConnector"}]}},
    {"kind":"standard_flow","businessKey":"Account_To_Remote","manageability":"defaults_only","payload":{"start":{"object":"Account"},"actionCalls":[{"name":"PushAccount","actionType":"PushToConnector"}]}},
    {"kind":"connector_field_mapping","businessKey":"jira-like-sample/push-mappings","manageability":"locked","payload":[
      {"sourceObject":"Opportunity","targetObject":"remoteDeal","actionType":"PushToConnector","direction":"outbound","mappings":[{"sourceField":"Name","targetField":"dealname"}]},
      {"sourceObject":"Account","targetObject":"remoteCompany","actionType":"PushToConnector","direction":"outbound","mappings":[{"sourceField":"Name","targetField":"name"}]}
    ]},
    {"kind":"connector_push_contract","businessKey":"jira-like-sample/Account_To_Remote/PushAccount/Account/remote-company","manageability":"locked","payload":{
      "flowDeveloperName":"Account_To_Remote",
      "flowActionName":"PushAccount",
      "slotLineageId":"account-remote-sync",
      "sourceObject":"Account",
      "destinationSlotId":"remote-company",
      "profileOwnerPack":"hubspot-connector",
      "profileName":"default",
      "targetObject":"remoteCompany",
      "createPathTemplate":"/crm/v3/objects/companies",
      "updatePathTemplate":"/crm/v3/objects/companies/{remoteId}",
      "responseIdPath":"id",
      "remoteIdField":"RemoteCompanyId__c"
    }},
    {"kind":"connector_push_contract","businessKey":"jira-like-sample/Opportunity_To_Remote/PushOpportunity/Opportunity/remote-deal","manageability":"locked","payload":{
      "flowDeveloperName":"Opportunity_To_Remote",
      "flowActionName":"PushOpportunity",
      "slotLineageId":"opportunity-remote-sync",
      "sourceObject":"Opportunity",
      "destinationSlotId":"remote-deal",
      "profileOwnerPack":"hubspot-connector",
      "profileName":"default",
      "targetObject":"remoteDeal",
      "createPathTemplate":"/crm/v3/objects/deals",
      "updatePathTemplate":"/crm/v3/objects/deals/{remoteId}",
      "responseIdPath":"id",
      "remoteIdField":"RemoteDealId__c",
      "associations":[{
        "sourceLookupField":"AccountId",
        "sourceObject":"Account",
        "remoteIdField":"RemoteCompanyId__c",
        "targetActionSlotId":"account-remote-sync",
        "targetDestinationSlotId":"remote-company",
        "targetObject":"remoteCompany",
        "payloadField":"associations",
        "payloadMergeMode":"append",
        "inlinePhases":["create"],
        "payloadTemplate":{"to":{"id":"{toId}"}}
      }]
    }}
  ]`, `"dependencies": [],`))
	writeFile(t, root, "objects/crm.json", `[
  {
    "apiName": "Issue",
    "label": "Issue",
    "pluralLabel": "Issues",
    "tableType": "standard_object",
    "category": "jira",
    "sharingModel": "Private",
    "enableHierarchySharing": false,
    "description": "Issue work item",
    "fields": [
      {"apiName": "Subject", "label": "Subject", "type": "string", "required": true, "isNameField": true},
      {"apiName": "PeerField__c", "label": "Peer Field", "type": "string"}
    ]
  },
  {
    "apiName": "Account",
    "label": "Account",
    "tableType": "standard_object",
    "category": "crm",
    "sharingModel": "Private",
    "enableHierarchySharing": false,
    "description": "Account",
    "fields": [
      {"apiName": "Name", "label": "Name", "type": "string", "required": true, "isNameField": true},
      {"apiName": "RemoteCompanyId__c", "label": "Remote Company ID", "type": "string"}
    ]
  },
  {
    "apiName": "Opportunity",
    "label": "Opportunity",
    "tableType": "standard_object",
    "category": "crm",
    "sharingModel": "Private",
    "enableHierarchySharing": false,
    "description": "Opportunity",
    "fields": [
      {"apiName": "Name", "label": "Name", "type": "string", "required": true, "isNameField": true},
      {"apiName": "RemoteDealId__c", "label": "Remote Deal ID", "type": "string"}
    ]
  }
]`)
	return root
}

func expectConnectorAssociationLookupSchemaLoad(mock sqlmock.Sqlmock) {
	mock.ExpectQuery("SELECT .*FROM `_System_Object`.*`IsDeleted` = \\?").
		WithArgs(false).
		WillReturnRows(crossPackSystemObjectRows().
			AddRow(crossPackSystemObjectRowValues("obj_opportunity", "Opportunity")...).
			AddRow(crossPackSystemObjectRowValues("obj_account", "Account")...))
	mock.ExpectQuery("SELECT .*FROM `_System_Field`.*`IsDeleted` = \\?").
		WithArgs(false).
		WillReturnRows(sqlmock.NewRows(constants.ColsSystemField).
			AddRow(crossPackSystemFieldRowValues("obj_opportunity", "AccountId", constants.FieldTypeReference, `["Account"]`)...).
			AddRow(crossPackSystemFieldRowValues("obj_account", "RemoteCompanyId__c", constants.FieldTypeString, nil)...))
	mock.ExpectQuery("SELECT .*FROM `_System_Relationship`").
		WillReturnRows(sqlmock.NewRows([]string{
			constants.FieldID,
			constants.FieldSysRelationship_ChildObjectAPIName,
			constants.FieldSysRelationship_ParentObjectAPIName,
			constants.FieldSysRelationship_FieldAPIName,
			constants.FieldSysRelationship_RelationshipName,
			constants.FieldSysRelationship_RelationshipType,
			constants.FieldSysRelationship_IsSharingParent,
			constants.FieldSysRelationship_CascadeDelete,
			constants.FieldSysRelationship_RestrictedDelete,
			constants.FieldSysRelationship_RelatedListLabel,
			constants.FieldSysRelationship_RelatedListFields,
			constants.FieldCreatedDate,
			constants.FieldLastModifiedDate,
		}))
}

func crossPackSystemObjectRows() *sqlmock.Rows {
	return sqlmock.NewRows(crossPackSystemObjectColumns())
}

func crossPackSystemObjectColumns() []string {
	cols := append([]string(nil), constants.ColsSystemObject...)
	return append(cols,
		constants.FieldSysObject_KeyPrefix,
		constants.FieldSysObject_DefaultChildRows,
		constants.FieldSysObject_DefaultChildRowsCustomized,
		constants.FieldSysObject_ChildCloneRulesCustomized,
		constants.FieldSysObject_ExclusiveSiblingFields,
		constants.FieldSysObject_ExclusiveSiblingFieldsCustomized,
		constants.FieldSysObject_ReadSource,
	)
}

func crossPackSystemObjectRowValues(id, apiName string) []driver.Value {
	values := []driver.Value{
		id,
		apiName,
		string(constants.TableTypeStandardObject),
		apiName,
		apiName + "s",
		nil,
		nil,
		nil,
		nil,
		false,
		false,
		true,
		string(constants.SharingModelPrivate),
		false,
		nil,
		nil,
		nil,
		nil,
		nil,
		nil,
		nil,
		nil,
		false,
		nil,
		false,
		nil,
		false,
		nil,
		nil,
		nil,
		nil,
		nil,
		nil,
		nil,
		false,
		false,
		nil,
		false,
		nil,
	}
	if len(values) != len(crossPackSystemObjectColumns()) {
		panic("crossPackSystemObjectRowValues out of sync with _System_Object columns")
	}
	return values
}

func crossPackSystemFieldRowValues(objectID, apiName string, fieldType constants.SchemaFieldType, referenceTo interface{}) []driver.Value {
	values := make([]driver.Value, len(constants.ColsSystemField))
	for i, col := range constants.ColsSystemField {
		switch col {
		case constants.FieldID:
			values[i] = "fld_" + strings.ToLower(apiName)
		case constants.FieldSysField_ObjectID:
			values[i] = objectID
		case constants.FieldSysField_APIName:
			values[i] = apiName
		case constants.FieldSysField_Label:
			values[i] = apiName
		case constants.FieldSysField_Type:
			values[i] = string(fieldType)
		case constants.FieldSysField_Createable, constants.FieldSysField_Updateable:
			values[i] = true
		case constants.FieldSysField_ReferenceTo:
			values[i] = referenceTo
		case constants.FieldSysField_OwnerKind:
			values[i] = constants.FieldOwnerKindRuntime
		case constants.FieldSysField_Required,
			constants.FieldSysField_IsUnique,
			constants.FieldSysField_IsSystem,
			constants.FieldSysField_IsNameField,
			constants.FieldSysField_Indexed,
			constants.FieldSysField_TrackHistory,
			constants.FieldSysField_IsMasterDetail,
			constants.FieldSysField_IsPolymorphic,
			constants.FieldIsDeleted,
			constants.FieldSysField_IsCustomized:
			values[i] = false
		default:
			values[i] = nil
		}
	}
	return values
}