package catalogapp

import (
	"context"
	"database/sql"
	"errors"
	"math"
	"strconv"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	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"
)

const (
	testIntegrationWorkload = "workload-integration"
	testPackDigest          = "nexus-pack-sha256-v1:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
)

func TestCanonicalIntegrationBindingSelectionsClosesFourLaunchKinds(t *testing.T) {
	values := []IntegrationBindingSelection{
		testIntegrationBindingSelection(consumerv1.BindingConnectorPushContract, "push"),
		testIntegrationBindingSelection(consumerv1.BindingConnectorInboundRoute, "route"),
		testIntegrationBindingSelection(consumerv1.BindingConnectorFieldMapping, "mapping"),
		testIntegrationBindingSelection(consumerv1.BindingConnectorCalloutProfile, "profile"),
	}
	values[3].Binding.CredentialRefs = []string{"credential-batch", "credential-primary"}
	values[1].Binding.RuntimeCommand = &consumerv1.RuntimeCommandBinding{
		CommandKind:          "runtime.flow.execute",
		PayloadFormatVersion: 1,
		DefinitionRef:        "FlowRoute",
		DefinitionVersion:    "flow-version-1",
	}
	got, err := canonicalIntegrationBindingSelections(
		values,
		[]string{testIntegrationWorkload},
		[]ConsumerGrant{testIntegrationConsumerGrant()},
	)
	if err != nil {
		t.Fatal(err)
	}
	wantKinds := []consumerv1.IntegrationBindingKind{
		consumerv1.BindingConnectorCalloutProfile,
		consumerv1.BindingConnectorFieldMapping,
		consumerv1.BindingConnectorInboundRoute,
		consumerv1.BindingConnectorPushContract,
	}
	if len(got) != len(wantKinds) {
		t.Fatalf("bindings = %d, want %d", len(got), len(wantKinds))
	}
	for index, kind := range wantKinds {
		if got[index].Binding.Kind != kind {
			t.Fatalf("binding[%d].Kind = %q, want %q", index, got[index].Binding.Kind, kind)
		}
	}
	if &got[0] == &values[0] {
		t.Fatal("canonicalization retained the caller slice")
	}
	route := got[2].Binding.RuntimeCommand
	if route == nil || route == values[1].Binding.RuntimeCommand ||
		route.DefinitionVersion != "flow-version-1" {
		t.Fatalf("Runtime command binding was not defensively cloned: %#v", route)
	}
}

func TestCanonicalIntegrationBindingSelectionsFailsClosed(t *testing.T) {
	valid := testIntegrationBindingSelection(
		consumerv1.BindingConnectorInboundRoute, "route",
	)
	for _, test := range []struct {
		name      string
		values    []IntegrationBindingSelection
		workloads []string
		grants    []ConsumerGrant
	}{
		{
			name:      "unauthorized workload",
			values:    []IntegrationBindingSelection{valid},
			workloads: []string{"workload-other"},
			grants:    []ConsumerGrant{testIntegrationConsumerGrant()},
		},
		{
			name:      "missing capability",
			values:    []IntegrationBindingSelection{valid},
			workloads: []string{testIntegrationWorkload},
			grants:    nil,
		},
		{
			name: "invalid pack digest",
			values: []IntegrationBindingSelection{func() IntegrationBindingSelection {
				value := valid
				value.Binding.PackPayloadDigest = "sha256:bad"
				return value
			}()},
			workloads: []string{testIntegrationWorkload},
			grants:    []ConsumerGrant{testIntegrationConsumerGrant()},
		},
		{
			name: "unsorted credentials",
			values: []IntegrationBindingSelection{func() IntegrationBindingSelection {
				value := valid
				value.Binding.CredentialRefs = []string{"z", "a"}
				return value
			}()},
			workloads: []string{testIntegrationWorkload},
			grants:    []ConsumerGrant{testIntegrationConsumerGrant()},
		},
		{
			name: "partial Runtime command",
			values: []IntegrationBindingSelection{func() IntegrationBindingSelection {
				value := valid
				value.Binding.RuntimeCommand = &consumerv1.RuntimeCommandBinding{
					CommandKind: "runtime.flow.execute",
				}
				return value
			}()},
			workloads: []string{testIntegrationWorkload},
			grants:    []ConsumerGrant{testIntegrationConsumerGrant()},
		},
		{
			name:      "duplicate natural identity",
			values:    []IntegrationBindingSelection{valid, valid},
			workloads: []string{testIntegrationWorkload},
			grants:    []ConsumerGrant{testIntegrationConsumerGrant()},
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			if _, err := canonicalIntegrationBindingSelections(
				test.values, test.workloads, test.grants,
			); err == nil {
				t.Fatal("invalid IntegrationBinding selection was accepted")
			}
		})
	}
}

func TestRuntimeCommandFromSQLFailsClosedOnPartialOrOverflow(t *testing.T) {
	valid := func(value string) sql.NullString {
		return sql.NullString{String: value, Valid: true}
	}
	tests := []struct {
		name        string
		kind        sql.NullString
		version     []byte
		definition  sql.NullString
		exact       sql.NullString
		wantNil     bool
		wantFailure bool
	}{
		{name: "N-1 row remains absent", wantNil: true},
		{
			name: "complete command",
			kind: valid("runtime.flow.execute"), version: []byte("1"),
			definition: valid("FlowRoute"), exact: valid("flow-version-1"),
		},
		{
			name: "partial columns",
			kind: valid("runtime.flow.execute"), version: []byte("1"),
			wantFailure: true,
		},
		{
			name:       "uint32 overflow",
			kind:       valid("runtime.flow.execute"),
			version:    []byte(strconv.FormatUint(uint64(math.MaxUint32)+1, 10)),
			definition: valid("FlowRoute"), exact: valid("flow-version-1"),
			wantFailure: true,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			got, err := runtimeCommandFromSQL(
				test.kind, test.version, test.definition, test.exact,
			)
			if (err != nil) != test.wantFailure {
				t.Fatalf("runtimeCommandFromSQL error = %v", err)
			}
			if test.wantNil && got != nil {
				t.Fatalf("runtimeCommandFromSQL = %#v, want nil", got)
			}
			if !test.wantNil && !test.wantFailure &&
				(got == nil || got.PayloadFormatVersion != 1 ||
					got.DefinitionVersion != "flow-version-1") {
				t.Fatalf("runtimeCommandFromSQL = %#v", got)
			}
		})
	}
}

func TestVerifyIntegrationPackAuthorityRequiresCompleteExactDefinitionSet(t *testing.T) {
	binding := testIntegrationBindingSelection(
		consumerv1.BindingConnectorInboundRoute,
		"route",
	).Binding

	for _, test := range []struct {
		name    string
		setMode string
		wantErr bool
	}{
		{name: "complete exact set", setMode: "complete"},
		{name: "missing set", setMode: "missing", wantErr: true},
		{name: "corrupt set digest", setMode: "corrupt", wantErr: true},
		{name: "binding absent from complete set", setMode: "empty", wantErr: true},
	} {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			mock.ExpectBegin()
			expectIntegrationPackGenerationAuthority(
				mock,
				binding,
				binding.PackInstallationID,
				nil,
				[]byte("2"),
				0,
			)
			expectIntegrationBindingDefinitionSet(t, mock, binding, test.setMode)
			mock.ExpectRollback()
			tx, err := db.BeginTx(context.Background(), nil)
			if err != nil {
				t.Fatal(err)
			}
			err = verifyIntegrationPackAuthorityTx(context.Background(), tx, binding)
			if test.wantErr != (err != nil) {
				t.Fatalf("authority error = %v, want error=%v", err, test.wantErr)
			}
			if test.wantErr && !errors.Is(err, ErrConflict) {
				t.Fatalf("authority error = %v, want ErrConflict", err)
			}
			_ = tx.Rollback()
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatal(err)
			}
		})
	}

	t.Run("binding is stale after replacement", func(t *testing.T) {
		db, mock, err := sqlmock.New()
		if err != nil {
			t.Fatal(err)
		}
		defer db.Close()
		mock.ExpectBegin()
		expectIntegrationPackGenerationAuthority(
			mock,
			binding,
			"install-pack-2",
			nil,
			[]byte("2"),
			0,
		)
		mock.ExpectRollback()
		tx, err := db.BeginTx(context.Background(), nil)
		if err != nil {
			t.Fatal(err)
		}
		err = verifyIntegrationPackAuthorityTx(context.Background(), tx, binding)
		if !errors.Is(err, ErrConflict) {
			t.Fatalf("stale binding error = %v, want ErrConflict", err)
		}
		_ = tx.Rollback()
		if err := mock.ExpectationsWereMet(); err != nil {
			t.Fatal(err)
		}
	})

	for _, test := range []struct {
		name              string
		previous          any
		activationEpoch   []byte
		activationDeleted int
	}{
		{name: "reserved previous pointer", previous: "install-pack-0", activationEpoch: []byte("2")},
		{name: "zero activation epoch", activationEpoch: []byte("0")},
		{name: "malformed activation epoch", activationEpoch: []byte("bad")},
		{name: "deleted activation", activationEpoch: []byte("2"), activationDeleted: 1},
	} {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			mock.ExpectBegin()
			expectIntegrationPackGenerationAuthority(
				mock,
				binding,
				binding.PackInstallationID,
				test.previous,
				test.activationEpoch,
				test.activationDeleted,
			)
			mock.ExpectRollback()
			tx, err := db.BeginTx(context.Background(), nil)
			if err != nil {
				t.Fatal(err)
			}
			err = verifyIntegrationPackAuthorityTx(context.Background(), tx, binding)
			if !errors.Is(err, ErrConflict) {
				t.Fatalf("malformed activation error = %v, want ErrConflict", err)
			}
			_ = tx.Rollback()
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatal(err)
			}
		})
	}
}

func TestEnsureIntegrationBindingSetRetryBackfillsOnlyLegacyEmptySelection(t *testing.T) {
	operation := Operation{
		OperationID:       "018f0000-0000-7000-9000-000000000001",
		AppInstallationID: "018f0000-0000-7000-9000-000000000002",
	}
	emptyDigest, err := ComputeIntegrationBindingSetDigest(nil)
	if err != nil {
		t.Fatal(err)
	}
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mock.ExpectBegin()
	mock.ExpectQuery("SELECT.*_System_CatalogIntegrationBindingSet").
		WithArgs(operation.OperationID).
		WillReturnError(sql.ErrNoRows)
	mock.ExpectExec("INSERT INTO _System_CatalogIntegrationBindingSet").
		WithArgs(
			operation.OperationID, operation.AppInstallationID, emptyDigest,
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectRollback()
	tx, err := db.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	if err := ensureIntegrationBindingSetForOperationRetryTx(
		context.Background(), tx, operation, nil,
	); err != nil {
		t.Fatal(err)
	}
	_ = tx.Rollback()
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}

	dbConflict, conflictMock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer dbConflict.Close()
	conflictMock.ExpectBegin()
	conflictMock.ExpectQuery("SELECT.*_System_CatalogIntegrationBindingSet").
		WithArgs(operation.OperationID).
		WillReturnError(sql.ErrNoRows)
	conflictMock.ExpectRollback()
	conflictTx, err := dbConflict.BeginTx(context.Background(), nil)
	if err != nil {
		t.Fatal(err)
	}
	err = ensureIntegrationBindingSetForOperationRetryTx(
		context.Background(),
		conflictTx,
		operation,
		[]IntegrationBindingSelection{testIntegrationBindingSelection(
			consumerv1.BindingConnectorInboundRoute, "route",
		)},
	)
	if !errors.Is(err, ErrConflict) {
		t.Fatalf("non-empty legacy backfill error = %v, want ErrConflict", err)
	}
	_ = conflictTx.Rollback()
	if err := conflictMock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestIntegrationBindingSetDigestCanonicalizesEmptySet(t *testing.T) {
	nilDigest, err := ComputeIntegrationBindingSetDigest(nil)
	if err != nil {
		t.Fatal(err)
	}
	emptyDigest, err := ComputeIntegrationBindingSetDigest(
		[]IntegrationBindingSelection{},
	)
	if err != nil {
		t.Fatal(err)
	}
	if nilDigest != emptyDigest {
		t.Fatalf("nil/empty IntegrationBinding digest = %s/%s", nilDigest, emptyDigest)
	}
}

func TestOperationRequestDigestPreservesEmptyBindingCompatibility(t *testing.T) {
	base := BeginOperationRequest{
		Kind: OperationPrepare, OrgID: "org-test", AppID: "io.example.app",
	}
	withNil, err := operationRequestDigest(base)
	if err != nil {
		t.Fatal(err)
	}
	base.IntegrationBindings = []IntegrationBindingSelection{}
	withEmpty, err := operationRequestDigest(base)
	if err != nil {
		t.Fatal(err)
	}
	if withNil != withEmpty {
		t.Fatalf("nil/empty binding request digest changed: %s != %s", withNil, withEmpty)
	}
	base.IntegrationBindings = nil
	base.ConsumerGrants = []ConsumerGrant{}
	withEmptyGrants, err := operationRequestDigest(base)
	if err != nil {
		t.Fatal(err)
	}
	if withNil != withEmptyGrants {
		t.Fatalf(
			"nil/empty consumer grant request digest changed: %s != %s",
			withNil, withEmptyGrants,
		)
	}
}

func testIntegrationBindingSelection(
	kind consumerv1.IntegrationBindingKind,
	businessKey string,
) IntegrationBindingSelection {
	return IntegrationBindingSelection{
		AuthorizedWorkloadID: testIntegrationWorkload,
		Binding: consumerv1.IntegrationBinding{
			Kind:               kind,
			OwnerPack:          "pack-integration",
			PackInstallationID: "install-pack-1",
			PackPayloadDigest:  testPackDigest,
			BusinessKey:        businessKey,
			DefinitionVersion:  1,
			CredentialRefs:     []string{},
		},
	}
}

func expectIntegrationPackGenerationAuthority(
	mock sqlmock.Sqlmock,
	binding consumerv1.IntegrationBinding,
	activePackInstallationID string,
	previous any,
	activationEpoch []byte,
	activationDeleted int,
) {
	mock.ExpectQuery("(?s)FROM _System_PackActivation AS a.*FOR UPDATE").
		WithArgs(binding.OwnerPack).
		WillReturnRows(sqlmock.NewRows([]string{
			"Id", "ActivePackInstallId", "PreviousPackInstallId", "ActivationEpoch",
			"ActivatedDate", "ActivationDeleted", "PackName", "PackInstallId",
			"Status", "DesiredPayloadHash", "AppliedPayloadHash", "GenerationDeleted",
		}).AddRow(
			connectorbindingdefinitionauthority.PackGenerationStorageID(
				binding.OwnerPack,
				activePackInstallationID,
			),
			activePackInstallationID,
			previous,
			activationEpoch,
			time.Now().UTC(),
			activationDeleted,
			binding.OwnerPack,
			activePackInstallationID,
			"active",
			binding.PackPayloadDigest,
			binding.PackPayloadDigest,
			0,
		))
}

func expectIntegrationBindingDefinitionSet(
	t *testing.T,
	mock sqlmock.Sqlmock,
	binding consumerv1.IntegrationBinding,
	mode string,
) {
	t.Helper()
	storageID := connectorbindingdefinitionauthority.PackGenerationStorageID(
		binding.OwnerPack,
		binding.PackInstallationID,
	)
	mock.ExpectQuery("(?s)FROM _System_PackGeneration.*WHERE Id = \\?.*AND BINARY Id = BINARY \\?.*LIMIT 1.*FOR UPDATE").
		WithArgs(storageID, storageID).
		WillReturnRows(sqlmock.NewRows([]string{
			"Id", "PackName", "PackInstallId", "AppliedPayloadHash",
		}).AddRow(
			storageID,
			binding.OwnerPack,
			binding.PackInstallationID,
			binding.PackPayloadDigest,
		))
	if mode == "missing" {
		mock.ExpectQuery("(?s)FROM _System_PackGenerationConnectorBindingDefinitionSet.*FOR UPDATE").
			WithArgs(storageID, storageID).
			WillReturnRows(sqlmock.NewRows([]string{
				"PackGenerationId", "Protocol", "DefinitionCount", "DefinitionSetDigest",
			}))
		return
	}
	definitions := []connectorbindingownerv1.DefinitionEnvelope(nil)
	if mode != "empty" {
		definition, err := connectorbindingownerv1.NewDefinitionEnvelope(
			connectorbindingownerv1.AuthoringInput{
				Identity:            connectorbindingownerv1.IdentityFromBinding(binding),
				CanonicalDefinition: []byte(`{"definition":"v1"}`),
			},
		)
		if err != nil {
			t.Fatal(err)
		}
		definitions = []connectorbindingownerv1.DefinitionEnvelope{definition}
	}
	set, err := connectorbindingownerv1.NewDefinitionSet(
		connectorbindingownerv1.DefinitionSetCoordinate{
			OwnerPack:          binding.OwnerPack,
			PackInstallationID: binding.PackInstallationID,
			PackPayloadDigest:  binding.PackPayloadDigest,
		},
		definitions,
	)
	if err != nil {
		t.Fatal(err)
	}
	setDigest := set.Header.OrderedSetDigest
	if mode == "corrupt" {
		setDigest = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	}
	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),
			setDigest,
		))
	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 testIntegrationConsumerGrant() ConsumerGrant {
	return ConsumerGrant{
		AuthorizedWorkloadID: testIntegrationWorkload,
		CapabilityID:         consumerv1.IntegrationBindingCapabilityName,
	}
}