package catalogreconcile

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"encoding/json"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	consumerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/connectorbindingdefinitionauthority"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

var connectorPackAuthorityColumns = []string{
	"Id", "ActivePackInstallId", "PreviousPackInstallId", "ActivationEpoch", "ActivatedDate", "ActivationDeleted",
	"PackName", "PackInstallId", "Status", "DesiredPayloadHash", "AppliedPayloadHash", "GenerationDeleted",
}

func TestResolveConnectorPushContractReadsOnlyExactImmutableGeneration(t *testing.T) {
	contact := validObject("Contact")
	contact.Fields = append(contact.Fields,
		models.FieldMetadata{ID: "field-contact-email", APIName: "Email", Label: "Email", Type: constants.FieldTypeString, PhysicalName: "email"},
		models.FieldMetadata{ID: "field-contact-remote", APIName: "RemoteId", Label: "Remote ID", Type: constants.FieldTypeString, PhysicalName: "remote_id"},
		models.FieldMetadata{ID: "field-contact-archived", APIName: "Archived", Label: "Archived", Type: constants.FieldTypeBoolean, PhysicalName: "archived"},
	)
	source := testConnectorPushSource()
	source.Manifest.Components[3].Payload = json.RawMessage(strings.Replace(
		string(source.Manifest.Components[3].Payload),
		`"sourceObject":"Contact",`,
		`"sourceObject":"Contact","recoverySourceEligibility":{"requiredFalseBooleanFields":["Archived"]},`,
		1,
	))
	pack, err := packs.NewSourceBackedPack(
		source,
		packs.WithConnectorPushValidationSchemaResolverForSource(
			source,
			packs.PackSourceOptions{ObjectSchemaResolver: func(apiName string) *models.ObjectMetadata {
				if apiName == "Contact" {
					return contact
				}
				return nil
			}},
		),
	)
	if err != nil {
		t.Fatalf("build source-backed connector pack: %v", err)
	}
	payloadDigest, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatal(err)
	}
	declaration, err := packs.BuildPersistedRuntimeDeclaration(pack)
	if err != nil {
		t.Fatal(err)
	}
	binding := testConnectorPushBinding()
	binding.PackPayloadDigest = payloadDigest
	binding.BusinessKey = "connector-a/Contact_To_Remote/PushContact/Contact/remote-contact"

	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := NewRepositoryMetadataReader(persistence.NewMetadataRepository(db))
	if err != nil {
		t.Fatal(err)
	}
	selected, err := resolveConnectorExecutionInTransaction(
		t,
		db,
		mock,
		true,
		func() {
			expectConnectorPackAuthority(t, mock, binding, nil, payloadDigest)
			expectGenerationRuntimeDeclaration(mock, binding, declaration.StartupShallowDeclaredRowsJSON())
		},
		func(txCtx context.Context) (ConnectorPushContractSelection, error) {
			return reader.ResolveConnectorPushContract(txCtx, binding)
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	if selected.Generation != binding.PackInstallationID || selected.Contract.ConnectorOwnerKey != "connector-a" ||
		selected.Contract.ActionSlotID != "contact-remote-sync" || selected.Contract.RemoteIDField != "RemoteId" ||
		selected.RecoverySourceEligibility == nil ||
		len(selected.RecoverySourceEligibility.RequiredFalseBooleanFields) != 1 ||
		selected.RecoverySourceEligibility.RequiredFalseBooleanFields[0] != "Archived" {
		t.Fatalf("resolved connector push selection = %#v", selected)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func TestResolveConnectorFieldMappingsReadsOnlyExactImmutableComponent(t *testing.T) {
	contact := validObject("Contact")
	contact.Fields = append(contact.Fields,
		models.FieldMetadata{ID: "field-contact-email", APIName: "Email", Label: "Email", Type: constants.FieldTypeString, PhysicalName: "email"},
		models.FieldMetadata{ID: "field-contact-remote", APIName: "RemoteId", Label: "Remote ID", Type: constants.FieldTypeString, PhysicalName: "remote_id"},
	)
	source := testConnectorPushSource()
	source.Manifest.Components[2].Payload = json.RawMessage(`[{"sourceObject":"Contact","targetObject":"remoteContact","actionType":"PushToConnector","direction":"bidirectional","mappings":[{"sourceField":"Email","targetField":"email"}]}]`)
	pack, err := packs.NewSourceBackedPack(
		source,
		packs.WithConnectorPushValidationSchemaResolverForSource(
			source,
			packs.PackSourceOptions{ObjectSchemaResolver: func(apiName string) *models.ObjectMetadata {
				if apiName == "Contact" {
					return contact
				}
				return nil
			}},
		),
	)
	if err != nil {
		t.Fatal(err)
	}
	digest, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatal(err)
	}
	declaration, err := packs.BuildPersistedRuntimeDeclaration(pack)
	if err != nil {
		t.Fatal(err)
	}
	binding := testConnectorPushBinding()
	binding.Kind = consumerv1.BindingConnectorFieldMapping
	binding.PackPayloadDigest = digest
	binding.BusinessKey = "connector-a/contact-push"
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := NewRepositoryMetadataReader(persistence.NewMetadataRepository(db))
	if err != nil {
		t.Fatal(err)
	}
	selected, err := resolveConnectorExecutionInTransaction(
		t,
		db,
		mock,
		true,
		func() {
			expectConnectorPackAuthority(t, mock, binding, nil, digest)
			expectGenerationRuntimeDeclaration(mock, binding, declaration.StartupShallowDeclaredRowsJSON())
		},
		func(txCtx context.Context) (ConnectorFieldMappingSelection, error) {
			return reader.ResolveConnectorFieldMappings(txCtx, binding)
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	if selected.Generation != binding.PackInstallationID || len(selected.Mappings) != 1 ||
		selected.Mappings[0].Direction != "inbound" || selected.Mappings[0].SourceField != "Email" ||
		len(selected.OutboundMappings) != 1 || selected.OutboundMappings[0].Direction != "bidirectional" {
		t.Fatalf("resolved exact connector mapping selection = %#v", selected)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestResolveConnectorCalloutProfileReadsOnlyExactAuthorizedGeneration(t *testing.T) {
	contact := validObject("Contact")
	contact.Fields = append(contact.Fields,
		models.FieldMetadata{ID: "field-contact-email", APIName: "Email", Label: "Email", Type: constants.FieldTypeString, PhysicalName: "email"},
		models.FieldMetadata{ID: "field-contact-remote", APIName: "RemoteId", Label: "Remote ID", Type: constants.FieldTypeString, PhysicalName: "remote_id"},
	)
	source := testConnectorPushSource()
	pack, err := packs.NewSourceBackedPack(
		source,
		packs.WithConnectorPushValidationSchemaResolverForSource(
			source,
			packs.PackSourceOptions{ObjectSchemaResolver: func(apiName string) *models.ObjectMetadata {
				if apiName == "Contact" {
					return contact
				}
				return nil
			}},
		),
	)
	if err != nil {
		t.Fatal(err)
	}
	digest, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatal(err)
	}
	declaration, err := packs.BuildPersistedRuntimeDeclaration(pack)
	if err != nil {
		t.Fatal(err)
	}
	binding := testConnectorPushBinding()
	binding.Kind = consumerv1.BindingConnectorCalloutProfile
	binding.PackPayloadDigest = digest
	binding.BusinessKey = "connector-a/default"
	binding.CredentialRefs = []string{"credential-a"}

	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := NewRepositoryMetadataReader(persistence.NewMetadataRepository(db))
	if err != nil {
		t.Fatal(err)
	}
	selected, err := resolveConnectorExecutionInTransaction(
		t,
		db,
		mock,
		true,
		func() {
			expectConnectorPackAuthority(t, mock, binding, nil, digest)
			expectGenerationRuntimeDeclaration(mock, binding, declaration.StartupShallowDeclaredRowsJSON())
		},
		func(txCtx context.Context) (ConnectorCalloutProfileSelection, error) {
			return reader.ResolveConnectorCalloutProfile(txCtx, binding)
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	if selected.Generation != binding.PackInstallationID ||
		selected.Profile.Generation != binding.PackInstallationID ||
		selected.Profile.CredentialRef != "credential-a" || selected.Profile.MaxAttempts != 3 ||
		len(selected.OutboundCapabilities) != 2 ||
		selected.OutboundCapabilities[0].TargetObject != "remoteContact" {
		t.Fatalf("resolved exact connector profile = %#v", selected)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestResolveConnectorPushContractFailsClosedOnStaleOrMixedAuthority(t *testing.T) {
	binding := testConnectorPushBinding()
	tests := []struct {
		name             string
		previous         driver.Value
		generationDigest string
		queryGeneration  bool
	}{
		{
			name:             "upgrade still has previous generation",
			previous:         "previous-install",
			generationDigest: binding.PackPayloadDigest,
		},
		{
			name:             "binding digest is stale",
			generationDigest: "sha256:new-generation",
		},
		{
			name:             "selected generation has no matching declaration",
			generationDigest: binding.PackPayloadDigest,
			queryGeneration:  true,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			reader, err := NewRepositoryMetadataReader(persistence.NewMetadataRepository(db))
			if err != nil {
				t.Fatal(err)
			}
			_, err = resolveConnectorExecutionInTransaction(
				t,
				db,
				mock,
				false,
				func() {
					expectConnectorPackAuthority(t, mock, binding, test.previous, test.generationDigest)
					if test.queryGeneration {
						mock.ExpectQuery("(?s)FROM _System_PackGeneration AS g.*FOR UPDATE").
							WithArgs(binding.OwnerPack, binding.PackInstallationID, binding.PackPayloadDigest).
							WillReturnRows(sqlmock.NewRows([]string{"RuntimeStartupShallowDeclarations"}))
					}
				},
				func(txCtx context.Context) (ConnectorPushContractSelection, error) {
					return reader.ResolveConnectorPushContract(txCtx, binding)
				},
			)
			if err == nil {
				t.Fatal("stale or mixed authority unexpectedly resolved")
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
		})
	}
}

func TestResolveConnectorPushContractRejectsMissingDefinitionSetBeforeRuntimeProjection(t *testing.T) {
	binding := testConnectorPushBinding()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := NewRepositoryMetadataReader(persistence.NewMetadataRepository(db))
	if err != nil {
		t.Fatal(err)
	}
	_, err = resolveConnectorExecutionInTransaction(
		t,
		db,
		mock,
		false,
		func() {
			expectConnectorPackGenerationAuthority(
				mock,
				binding,
				nil,
				binding.PackPayloadDigest,
			)
			expectConnectorBindingDefinitionSet(t, mock, binding, true)
		},
		func(txCtx context.Context) (ConnectorPushContractSelection, error) {
			return reader.ResolveConnectorPushContract(txCtx, binding)
		},
	)
	if err == nil || !strings.Contains(err.Error(), "missing its connector binding definition set") {
		t.Fatalf("missing definition-set error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func testConnectorPushSource() *packs.ResolvedPackSource {
	const (
		flow     = `{"start":{"object":"Contact"},"actionCalls":[{"name":"PushContact","actionType":"PushToConnector"}]}`
		mapping  = `[{"sourceObject":"Contact","targetObject":"remoteContact","actionType":"PushToConnector","direction":"outbound","mappings":[{"sourceField":"Email","targetField":"email"}]}]`
		contract = `{"flowDeveloperName":"Contact_To_Remote","flowActionName":"PushContact","slotLineageId":"contact-remote-sync","sourceObject":"Contact","destinationSlotId":"remote-contact","profileOwnerPack":"connector-a","profileName":"default","targetObject":"remoteContact","createPathTemplate":"/contacts","updatePathTemplate":"/contacts/{remoteId}","responseIdPath":"id","remoteIdField":"RemoteId"}`
		profile  = `{"profileName":"default","credentialRef":"credential-a","endpointClass":"crm.write","retryPolicy":{"maxAttempts":3,"baseDelay":1000000,"multiplier":2,"maxDelay":10000000,"retryableStatusCodes":[503,429]},"outboundCapabilities":[{"operation":"create","targetObject":"remoteContact","method":"POST","pathTemplate":"/contacts"},{"operation":"update","targetObject":"remoteContact","method":"PATCH","pathTemplate":"/contacts/{remoteId}"}]}`
	)
	return &packs.ResolvedPackSource{
		Kind:     packs.PackSourceKindOfficialRepo,
		URI:      "connector-a",
		Revision: "source-revision-a",
		Manifest: &packs.PackManifest{
			ManifestFormatVersion: packs.SupportedManifestFormatVersion,
			Name:                  "connector-a", DisplayName: "Connector A", Category: packs.ConnectorPack, Version: "1.0.0",
			PlatformCompatibility: packs.PlatformCompatibility{Min: "0.1.0", MaxExclusive: "2.0.0"},
			Components: []packs.ManifestComponent{
				{Kind: packs.ComponentKindStandardFlow, BusinessKey: "Contact_To_Remote", Manageability: packs.ManageabilityDefaultsOnly, Payload: json.RawMessage(flow)},
				{Kind: packs.ComponentKindConnectorCalloutProfile, BusinessKey: "connector-a/default", Manageability: packs.ManageabilityLocked, Payload: json.RawMessage(profile)},
				{Kind: packs.ComponentKindConnectorFieldMapping, BusinessKey: "connector-a/contact-push", Manageability: packs.ManageabilityLocked, Payload: json.RawMessage(mapping)},
				{Kind: packs.ComponentKindConnectorPushContract, BusinessKey: "connector-a/Contact_To_Remote/PushContact/Contact/remote-contact", Manageability: packs.ManageabilityLocked, Payload: json.RawMessage(contract)},
			},
		},
	}
}

func expectConnectorPackAuthority(
	t *testing.T,
	mock sqlmock.Sqlmock,
	binding consumerv1.IntegrationBinding,
	previous driver.Value,
	generationDigest string,
) {
	t.Helper()
	expectConnectorPackGenerationAuthority(
		mock,
		binding,
		previous,
		generationDigest,
	)
	if previous == nil && generationDigest == binding.PackPayloadDigest {
		expectConnectorBindingDefinitionSet(t, mock, binding, false)
	}
}

func expectConnectorPackGenerationAuthority(
	mock sqlmock.Sqlmock,
	binding consumerv1.IntegrationBinding,
	previous driver.Value,
	generationDigest string,
) {
	storageID := connectorbindingdefinitionauthority.PackGenerationStorageID(
		binding.OwnerPack,
		binding.PackInstallationID,
	)
	mock.ExpectQuery("(?s)FROM _System_PackActivation AS a.*FOR UPDATE").
		WithArgs(binding.OwnerPack).
		WillReturnRows(sqlmock.NewRows(connectorPackAuthorityColumns).AddRow(
			storageID,
			binding.PackInstallationID,
			previous,
			[]byte("17"),
			time.Now(),
			0,
			binding.OwnerPack,
			binding.PackInstallationID,
			"active",
			generationDigest,
			generationDigest,
			0,
		))
}

func expectConnectorBindingDefinitionSet(
	t *testing.T,
	mock sqlmock.Sqlmock,
	binding consumerv1.IntegrationBinding,
	missing bool,
) {
	t.Helper()
	storageID := connectorbindingdefinitionauthority.PackGenerationStorageID(
		binding.OwnerPack,
		binding.PackInstallationID,
	)
	mock.ExpectQuery("(?s)FROM _System_PackGeneration.*WHERE Id = \\?.*AND BINARY Id = BINARY \\?.*FOR UPDATE").
		WithArgs(storageID, storageID).
		WillReturnRows(sqlmock.NewRows([]string{
			"Id", "PackName", "PackInstallId", "AppliedPayloadHash",
		}).AddRow(
			storageID,
			binding.OwnerPack,
			binding.PackInstallationID,
			binding.PackPayloadDigest,
		))
	if missing {
		mock.ExpectQuery("(?s)FROM _System_PackGenerationConnectorBindingDefinitionSet.*FOR UPDATE").
			WithArgs(storageID, storageID).
			WillReturnRows(sqlmock.NewRows([]string{
				"PackGenerationId", "Protocol", "DefinitionCount", "DefinitionSetDigest",
			}))
		return
	}
	canonicalDefinition := []byte(`{"definition":"v1"}`)
	if binding.Kind == consumerv1.BindingConnectorCalloutProfile {
		canonicalDefinition = []byte(`{"credentialRef":"credential-a","endpointClass":"crm.write","outboundCapabilities":[{"method":"POST","operation":"create","pathTemplate":"/contacts","targetObject":"remoteContact"},{"method":"PATCH","operation":"update","pathTemplate":"/contacts/{remoteId}","targetObject":"remoteContact"}],"profileName":"default","retryPolicy":{"baseDelay":1000000,"maxAttempts":3,"maxDelay":10000000,"multiplier":2,"retryableStatusCodes":[503,429]}}`)
	}
	definition, err := connectorbindingownerv1.NewDefinitionEnvelope(
		connectorbindingownerv1.AuthoringInput{
			Identity:            connectorbindingownerv1.IdentityFromBinding(binding),
			CanonicalDefinition: canonicalDefinition,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	set, err := connectorbindingownerv1.NewDefinitionSet(
		connectorbindingownerv1.DefinitionSetCoordinate{
			OwnerPack:          binding.OwnerPack,
			PackInstallationID: binding.PackInstallationID,
			PackPayloadDigest:  binding.PackPayloadDigest,
		},
		[]connectorbindingownerv1.DefinitionEnvelope{definition},
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectQuery("(?s)FROM _System_PackGenerationConnectorBindingDefinitionSet.*FOR UPDATE").
		WithArgs(storageID, storageID).
		WillReturnRows(sqlmock.NewRows([]string{
			"PackGenerationId", "Protocol", "DefinitionCount", "DefinitionSetDigest",
		}).AddRow(
			storageID,
			set.Header.Protocol,
			strconv.FormatUint(set.Header.DefinitionCount, 10),
			set.Header.OrderedSetDigest,
		))
	rows := sqlmock.NewRows([]string{
		"Id", "PackGenerationId", "PackInstallId", "DefinitionOrdinal",
		"Kind", "OwnerPack", "PackPayloadDigest", "BusinessKey",
		"DefinitionVersion", "DefinitionEnvelopeDigest", "DefinitionEnvelope",
	})
	for _, entry := range set.Entries {
		rows.AddRow(
			connectorbindingdefinitionauthority.DefinitionStorageID(
				storageID,
				entry.Kind,
				entry.BusinessKey,
			),
			storageID,
			entry.PackInstallationID,
			strconv.FormatUint(entry.Ordinal, 10),
			entry.Kind,
			entry.OwnerPack,
			entry.PackPayloadDigest,
			entry.BusinessKey,
			strconv.FormatUint(entry.DefinitionVersion, 10),
			entry.EnvelopeDigest,
			entry.EnvelopeJSON,
		)
	}
	mock.ExpectQuery("(?s)FROM _System_PackGenerationConnectorBindingDefinition.*ORDER BY DefinitionOrdinal.*FOR UPDATE").
		WithArgs(storageID).
		WillReturnRows(rows)
}

func expectGenerationRuntimeDeclaration(
	mock sqlmock.Sqlmock,
	binding consumerv1.IntegrationBinding,
	declaration string,
) {
	mock.ExpectQuery("(?s)FROM _System_PackGeneration AS g.*FOR UPDATE").
		WithArgs(binding.OwnerPack, binding.PackInstallationID, binding.PackPayloadDigest).
		WillReturnRows(sqlmock.NewRows([]string{"RuntimeStartupShallowDeclarations"}).AddRow(declaration))
}

func resolveConnectorExecutionInTransaction[T any](
	t *testing.T,
	db *sql.DB,
	mock sqlmock.Sqlmock,
	expectCommit bool,
	expectQueries func(),
	resolve func(context.Context) (T, error),
) (T, error) {
	t.Helper()
	mock.ExpectBegin()
	expectQueries()
	if expectCommit {
		mock.ExpectCommit()
	} else {
		mock.ExpectRollback()
	}
	manager := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	var selected T
	err := manager.WithTransaction(context.Background(), func(txCtx context.Context) error {
		var err error
		selected, err = resolve(txCtx)
		return err
	})
	return selected, err
}

func testConnectorPushBinding() consumerv1.IntegrationBinding {
	return consumerv1.IntegrationBinding{
		Kind:               consumerv1.BindingConnectorPushContract,
		OwnerPack:          "connector-a",
		PackInstallationID: "install-a",
		PackPayloadDigest:  "nexus-pack-sha256-v1:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
		BusinessKey:        "push/account",
		DefinitionVersion:  1,
	}
}

func TestResolveConnectorPushContractRejectsMissingPersistedProjection(t *testing.T) {
	binding := testConnectorPushBinding()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := NewRepositoryMetadataReader(persistence.NewMetadataRepository(db))
	if err != nil {
		t.Fatal(err)
	}
	_, err = resolveConnectorExecutionInTransaction(
		t,
		db,
		mock,
		false,
		func() {
			expectConnectorPackAuthority(t, mock, binding, nil, binding.PackPayloadDigest)
			expectGenerationRuntimeDeclaration(mock, binding, `[]`)
		},
		func(txCtx context.Context) (ConnectorPushContractSelection, error) {
			return reader.ResolveConnectorPushContract(txCtx, binding)
		},
	)
	if err == nil || !strings.Contains(err.Error(), "absent from persisted generation") {
		t.Fatalf("missing projection error = %v", err)
	}
}