package catalogapp

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"slices"
	"sort"
	"strconv"

	consumerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogconsumerauthority"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/connectorbindingdefinitionauthority"
)

const integrationBindingGenerationActive = "active"

func canonicalIntegrationBindingSelections(
	values []IntegrationBindingSelection,
	authorizedWorkloads []string,
	consumerGrants []ConsumerGrant,
) ([]IntegrationBindingSelection, error) {
	result := make([]IntegrationBindingSelection, len(values))
	for index, value := range values {
		result[index] = IntegrationBindingSelection{
			AuthorizedWorkloadID: value.AuthorizedWorkloadID,
			Binding: consumerv1.IntegrationBinding{
				Kind:               value.Binding.Kind,
				OwnerPack:          value.Binding.OwnerPack,
				PackInstallationID: value.Binding.PackInstallationID,
				PackPayloadDigest:  value.Binding.PackPayloadDigest,
				BusinessKey:        value.Binding.BusinessKey,
				DefinitionVersion:  value.Binding.DefinitionVersion,
				CredentialRefs:     append([]string(nil), value.Binding.CredentialRefs...),
				RuntimeCommand:     cloneRuntimeCommandBinding(value.Binding.RuntimeCommand),
			},
		}
		if err := ValidateIntegrationBindingSelection(result[index]); err != nil {
			return nil, fmt.Errorf("integration binding[%d]: %w", index, err)
		}
		if !slices.Contains(authorizedWorkloads, result[index].AuthorizedWorkloadID) {
			return nil, fmt.Errorf(
				"integration binding[%d] workload %q is not authorized by the App operation",
				index, result[index].AuthorizedWorkloadID,
			)
		}
		if !hasConsumerGrant(
			consumerGrants,
			result[index].AuthorizedWorkloadID,
			consumerv1.IntegrationBindingCapabilityName,
		) {
			return nil, fmt.Errorf(
				"integration binding[%d] workload %q lacks its Catalog consumer grant",
				index, result[index].AuthorizedWorkloadID,
			)
		}
	}
	sort.Slice(result, func(left, right int) bool {
		return integrationBindingSelectionKey(result[left]) <
			integrationBindingSelectionKey(result[right])
	})
	for index := 1; index < len(result); index++ {
		if integrationBindingSelectionKey(result[index-1]) ==
			integrationBindingSelectionKey(result[index]) {
			return nil, fmt.Errorf("integration bindings contain a duplicate natural identity")
		}
	}
	return result, nil
}

// ValidateIntegrationBindingSelection validates one neutral exact-generation
// reference without assigning workload or pack-generation authority.
func ValidateIntegrationBindingSelection(value IntegrationBindingSelection) error {
	return catalogconsumerauthority.ValidateIntegrationBindingSelection(value)
}

func integrationBindingSelectionKey(value IntegrationBindingSelection) string {
	binding := value.Binding
	return value.AuthorizedWorkloadID + "\x00" + string(binding.Kind) + "\x00" +
		binding.OwnerPack + "\x00" + binding.PackInstallationID + "\x00" +
		binding.BusinessKey
}

// ComputeIntegrationBindingSetDigest returns the durable commitment over one
// canonical operation-bound binding selection. Callers must supply the exact
// sorted selection order; lifecycle admission owns shape validation.
func ComputeIntegrationBindingSetDigest(values []IntegrationBindingSelection) (string, error) {
	return catalogconsumerauthority.ComputeIntegrationBindingSetDigest(values)
}

func stageIntegrationBindingsTx(
	ctx context.Context,
	tx *sql.Tx,
	operationID string,
	installationID string,
	orgID string,
	values []IntegrationBindingSelection,
) error {
	for _, value := range values {
		if err := verifyIntegrationPackAuthorityTx(ctx, tx, value.Binding); err != nil {
			return err
		}
		credentials, err := json.Marshal(value.Binding.CredentialRefs)
		if err != nil {
			return err
		}
		commandKind, commandPayloadVersion, commandDefinitionRef,
			commandDefinitionVersion := runtimeCommandSQLValues(
			value.Binding.RuntimeCommand,
		)
		bindingIdentity := []string{
			"integration-binding", operationID, value.AuthorizedWorkloadID,
			string(value.Binding.Kind), value.Binding.OwnerPack,
			value.Binding.PackInstallationID, value.Binding.PackPayloadDigest,
			value.Binding.BusinessKey,
			strconv.FormatUint(value.Binding.DefinitionVersion, 10),
			string(credentials),
		}
		if value.Binding.RuntimeCommand != nil {
			runtimeCommand, err := json.Marshal(value.Binding.RuntimeCommand)
			if err != nil {
				return err
			}
			bindingIdentity = append(bindingIdentity, string(runtimeCommand))
		}
		bindingID := hashKey(bindingIdentity...)
		_, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogIntegrationBinding
(IntegrationBindingId, OrgId, AppInstallationId, OperationId,
AuthorizedWorkloadId, Kind, OwnerPack, PackInstallId, PackPayloadDigest,
BusinessKey, DefinitionVersion, CredentialRefs, CommandKind,
CommandPayloadFormatVersion, CommandDefinitionRef, CommandDefinitionVersion,
CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?, ?, ?,
CURRENT_TIMESTAMP(6))`,
			bindingID, orgID, installationID, operationID,
			value.AuthorizedWorkloadID, value.Binding.Kind, value.Binding.OwnerPack,
			value.Binding.PackInstallationID, value.Binding.PackPayloadDigest,
			value.Binding.BusinessKey, epochText(value.Binding.DefinitionVersion),
			credentials, commandKind, commandPayloadVersion,
			commandDefinitionRef, commandDefinitionVersion,
		)
		if isDuplicate(err) {
			return ErrConflict
		}
		if err != nil {
			return err
		}
	}
	setDigest, err := ComputeIntegrationBindingSetDigest(values)
	if err != nil {
		return err
	}
	_, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogIntegrationBindingSet
(OperationId, AppInstallationId, BindingSetDigest, BindingCount, CreatedDate)
VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
		operationID, installationID, setDigest, epochText(uint64(len(values))),
	)
	if isDuplicate(err) {
		return ErrConflict
	}
	return err
}

func ensureIntegrationBindingSetForOperationRetryTx(
	ctx context.Context,
	tx *sql.Tx,
	operation Operation,
	values []IntegrationBindingSelection,
) error {
	expectedDigest, err := ComputeIntegrationBindingSetDigest(values)
	if err != nil {
		return err
	}
	var (
		installationID string
		storedDigest   string
		rawCount       []byte
	)
	err = tx.QueryRowContext(ctx, `SELECT
AppInstallationId, BindingSetDigest, BindingCount
FROM _System_CatalogIntegrationBindingSet
WHERE OperationId = ?
FOR UPDATE`, operation.OperationID).Scan(
		&installationID, &storedDigest, &rawCount,
	)
	if errors.Is(err, sql.ErrNoRows) {
		// Empty was omitted from the v1 request JSON, so an exact pre-M1
		// operation retry has the same request digest. Backfilling only that
		// empty commitment preserves N/N-1 without inventing a binding.
		if len(values) != 0 {
			return fmt.Errorf(
				"%w: non-empty IntegrationBinding selection has no durable commitment",
				ErrConflict,
			)
		}
		_, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogIntegrationBindingSet
(OperationId, AppInstallationId, BindingSetDigest, BindingCount, CreatedDate)
VALUES (?, ?, ?, 0, CURRENT_TIMESTAMP(6))`,
			operation.OperationID, operation.AppInstallationID, expectedDigest,
		)
		if isDuplicate(err) {
			return ErrConflict
		}
		return err
	}
	if err != nil {
		return err
	}
	count, err := parseEpochAllowZero(rawCount)
	if err != nil {
		return err
	}
	if installationID != operation.AppInstallationID ||
		storedDigest != expectedDigest ||
		count != uint64(len(values)) {
		return fmt.Errorf(
			"%w: IntegrationBinding commitment disagrees with exact operation retry",
			ErrConflict,
		)
	}
	return nil
}

type IntegrationPackAuthorityQueryer interface {
	QueryRowContext(context.Context, string, ...interface{}) *sql.Row
}

// VerifyIntegrationPackAuthority reads and locks the exact active immutable
// generation and its complete source-free definition set. Callers compose it
// inside their existing Catalog transaction; it does not resolve source files.
func VerifyIntegrationPackAuthority(
	ctx context.Context,
	queryer IntegrationPackAuthorityQueryer,
	binding consumerv1.IntegrationBinding,
) error {
	_, err := readIntegrationPackAuthorityDefinition(ctx, queryer, binding)
	return err
}

// ReadIntegrationPackAuthorityDefinition verifies the same exact active
// generation and complete owner definition set as VerifyIntegrationPackAuthority
// and returns the selected canonical definition bytes. Catalog compilers use
// this when compatibility depends on provider-authored definition content, not
// only on the binding identity.
func ReadIntegrationPackAuthorityDefinition(
	ctx context.Context,
	queryer IntegrationPackAuthorityQueryer,
	binding consumerv1.IntegrationBinding,
) ([]byte, error) {
	return readIntegrationPackAuthorityDefinition(ctx, queryer, binding)
}

func readIntegrationPackAuthorityDefinition(
	ctx context.Context,
	queryer IntegrationPackAuthorityQueryer,
	binding consumerv1.IntegrationBinding,
) ([]byte, error) {
	reader, ok := queryer.(connectorbindingdefinitionauthority.Reader)
	if !ok {
		return nil, fmt.Errorf(
			"%w: integration binding authority reader cannot verify a complete definition set",
			ErrConflict,
		)
	}
	var (
		generationStorageID  string
		activePackInstallID  string
		previousPackInstall  sql.NullString
		rawActivationEpoch   []byte
		activatedDate        sql.NullTime
		activationDeleted    int
		packName             string
		packInstallID        string
		status               string
		desiredPayloadDigest string
		appliedPayloadDigest string
		generationDeleted    int
	)
	err := queryer.QueryRowContext(ctx, `
SELECT
  g.Id,
  a.ActivePackInstallId,
  a.PreviousPackInstallId,
  a.ActivationEpoch,
  a.ActivatedDate,
  COALESCE(a.IsDeleted, 0),
  g.PackName,
  g.PackInstallId,
  g.Status,
  g.DesiredPayloadHash,
  g.AppliedPayloadHash,
  COALESCE(g.IsDeleted, 0)
FROM _System_PackActivation AS a
JOIN _System_PackGeneration AS g
  ON g.PackInstallId = a.ActivePackInstallId
WHERE a.PackName = ?
LIMIT 1
FOR UPDATE`, binding.OwnerPack).Scan(
		&generationStorageID,
		&activePackInstallID,
		&previousPackInstall,
		&rawActivationEpoch,
		&activatedDate,
		&activationDeleted,
		&packName,
		&packInstallID,
		&status,
		&desiredPayloadDigest,
		&appliedPayloadDigest,
		&generationDeleted,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, fmt.Errorf(
			"%w: integration binding pack %q has no active immutable generation",
			ErrNotReady,
			binding.OwnerPack,
		)
	}
	if err != nil {
		return nil, err
	}
	activationEpoch, err := parseEpoch(rawActivationEpoch)
	if err != nil {
		return nil, fmt.Errorf(
			"%w: integration binding pack %q has malformed activation authority",
			ErrConflict,
			binding.OwnerPack,
		)
	}
	if activePackInstallID != binding.PackInstallationID ||
		previousPackInstall.Valid ||
		activationEpoch == 0 ||
		!activatedDate.Valid ||
		activationDeleted != 0 ||
		packName != binding.OwnerPack ||
		packInstallID != binding.PackInstallationID ||
		status != integrationBindingGenerationActive ||
		desiredPayloadDigest != appliedPayloadDigest ||
		appliedPayloadDigest != binding.PackPayloadDigest ||
		generationDeleted != 0 {
		return nil, fmt.Errorf(
			"%w: integration binding %s/%s does not match the exact active pack generation",
			ErrConflict,
			binding.OwnerPack,
			binding.BusinessKey,
		)
	}
	definition, err := connectorbindingdefinitionauthority.ReadVerifiedBindingDefinition(
		ctx,
		reader,
		connectorbindingdefinitionauthority.GenerationCoordinate{
			StorageID:          generationStorageID,
			OwnerPack:          packName,
			PackInstallationID: packInstallID,
			PackPayloadDigest:  appliedPayloadDigest,
		},
		binding,
	)
	if errors.Is(err, connectorbindingdefinitionauthority.ErrPrecondition) ||
		errors.Is(err, connectorbindingdefinitionauthority.ErrConflict) {
		return nil, fmt.Errorf(
			"%w: integration binding %s/%s owner authority: %v",
			ErrConflict,
			binding.OwnerPack,
			binding.BusinessKey,
			err,
		)
	}
	if err != nil {
		return nil, err
	}
	return append([]byte(nil), definition.CanonicalDefinition...), nil
}

func verifyIntegrationPackAuthorityTx(
	ctx context.Context,
	tx *sql.Tx,
	binding consumerv1.IntegrationBinding,
) error {
	return VerifyIntegrationPackAuthority(ctx, tx, binding)
}

func readOperationIntegrationBindingsTx(
	ctx context.Context,
	tx *sql.Tx,
	operation Operation,
) ([]IntegrationBindingSelection, error) {
	rows, err := tx.QueryContext(ctx, `SELECT
AuthorizedWorkloadId, Kind, OwnerPack, PackInstallId, PackPayloadDigest,
BusinessKey, DefinitionVersion, CredentialRefs, CommandKind,
CommandPayloadFormatVersion, CommandDefinitionRef, CommandDefinitionVersion,
OrgId, AppInstallationId
FROM _System_CatalogIntegrationBinding
WHERE OperationId = ?
ORDER BY BINARY AuthorizedWorkloadId, BINARY Kind, BINARY OwnerPack,
         BINARY PackInstallId, BINARY BusinessKey
FOR UPDATE`, operation.OperationID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var result []IntegrationBindingSelection
	for rows.Next() {
		var (
			value             IntegrationBindingSelection
			rawKind           string
			rawVersion        []byte
			rawCredentialRefs []byte
			commandKind       sql.NullString
			rawCommandVersion []byte
			commandDefinitionRef,
			commandDefinitionVersion sql.NullString
			orgID             string
			appInstallationID string
		)
		if err := rows.Scan(
			&value.AuthorizedWorkloadID,
			&rawKind,
			&value.Binding.OwnerPack,
			&value.Binding.PackInstallationID,
			&value.Binding.PackPayloadDigest,
			&value.Binding.BusinessKey,
			&rawVersion,
			&rawCredentialRefs,
			&commandKind,
			&rawCommandVersion,
			&commandDefinitionRef,
			&commandDefinitionVersion,
			&orgID,
			&appInstallationID,
		); err != nil {
			return nil, err
		}
		if orgID != operation.OrgID || appInstallationID != operation.AppInstallationID {
			return nil, fmt.Errorf(
				"%w: IntegrationBinding disagrees with its operation",
				ErrConflict,
			)
		}
		value.Binding.Kind = consumerv1.IntegrationBindingKind(rawKind)
		value.Binding.DefinitionVersion, err = parseEpoch(rawVersion)
		if err != nil {
			return nil, err
		}
		if err := json.Unmarshal(rawCredentialRefs, &value.Binding.CredentialRefs); err != nil {
			return nil, fmt.Errorf(
				"%w: decode operation IntegrationBinding credential references: %v",
				ErrCorrupt, err,
			)
		}
		value.Binding.RuntimeCommand, err = runtimeCommandFromSQL(
			commandKind,
			rawCommandVersion,
			commandDefinitionRef,
			commandDefinitionVersion,
		)
		if err != nil {
			return nil, fmt.Errorf(
				"%w: decode operation IntegrationBinding Runtime command: %v",
				ErrCorrupt, err,
			)
		}
		if err := ValidateIntegrationBindingSelection(value); err != nil {
			return nil, fmt.Errorf("%w: invalid operation IntegrationBinding: %v", ErrCorrupt, err)
		}
		result = append(result, value)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return result, nil
}

func cloneRuntimeCommandBinding(
	value *consumerv1.RuntimeCommandBinding,
) *consumerv1.RuntimeCommandBinding {
	if value == nil {
		return nil
	}
	cloned := *value
	return &cloned
}

func runtimeCommandSQLValues(
	value *consumerv1.RuntimeCommandBinding,
) (any, any, any, any) {
	if value == nil {
		return nil, nil, nil, nil
	}
	return value.CommandKind,
		epochText(uint64(value.PayloadFormatVersion)),
		value.DefinitionRef,
		value.DefinitionVersion
}

func runtimeCommandFromSQL(
	commandKind sql.NullString,
	rawPayloadFormatVersion []byte,
	definitionRef sql.NullString,
	definitionVersion sql.NullString,
) (*consumerv1.RuntimeCommandBinding, error) {
	present := 0
	for _, isPresent := range []bool{
		commandKind.Valid,
		len(rawPayloadFormatVersion) != 0,
		definitionRef.Valid,
		definitionVersion.Valid,
	} {
		if isPresent {
			present++
		}
	}
	if present == 0 {
		return nil, nil
	}
	if present != 4 {
		return nil, errors.New("Runtime command columns are partially populated")
	}
	payloadFormatVersion, err := parseEpoch(rawPayloadFormatVersion)
	if err != nil || payloadFormatVersion > math.MaxUint32 {
		return nil, errors.New("Runtime command payload format version is invalid")
	}
	return &consumerv1.RuntimeCommandBinding{
		CommandKind:          commandKind.String,
		PayloadFormatVersion: uint32(payloadFormatVersion),
		DefinitionRef:        definitionRef.String,
		DefinitionVersion:    definitionVersion.String,
	}, nil
}

func verifyOperationIntegrationBindingsTx(
	ctx context.Context,
	tx *sql.Tx,
	operation Operation,
) error {
	values, err := readOperationIntegrationBindingsTx(ctx, tx, operation)
	if err != nil {
		return err
	}
	authorizedWorkloads, err := readOperationAuthorizedWorkloadsTx(
		ctx, tx, operation,
	)
	if err != nil {
		return err
	}
	consumerGrants, err := readOperationConsumerGrantsTx(ctx, tx, operation)
	if err != nil {
		return err
	}
	canonical, err := canonicalIntegrationBindingSelections(
		values,
		authorizedWorkloads,
		consumerGrants,
	)
	if err != nil {
		return fmt.Errorf("%w: operation IntegrationBinding set is invalid: %v", ErrCorrupt, err)
	}
	if !slices.EqualFunc(values, canonical, func(
		left IntegrationBindingSelection,
		right IntegrationBindingSelection,
	) bool {
		return integrationBindingSelectionKey(left) == integrationBindingSelectionKey(right) &&
			left.Binding.PackPayloadDigest == right.Binding.PackPayloadDigest &&
			left.Binding.DefinitionVersion == right.Binding.DefinitionVersion &&
			slices.Equal(left.Binding.CredentialRefs, right.Binding.CredentialRefs)
	}) {
		return fmt.Errorf("%w: operation IntegrationBinding set is not canonical", ErrCorrupt)
	}
	for _, value := range values {
		if err := verifyIntegrationPackAuthorityTx(ctx, tx, value.Binding); err != nil {
			return err
		}
	}
	var (
		storedInstallationID string
		storedDigest         string
		rawCount             []byte
	)
	err = tx.QueryRowContext(ctx, `SELECT AppInstallationId, BindingSetDigest, BindingCount
FROM _System_CatalogIntegrationBindingSet
WHERE OperationId = ?
FOR UPDATE`, operation.OperationID).Scan(
		&storedInstallationID, &storedDigest, &rawCount,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf("%w: operation has no IntegrationBinding set commitment", ErrCorrupt)
	}
	if err != nil {
		return err
	}
	count, err := parseEpochAllowZero(rawCount)
	if err != nil {
		return err
	}
	currentDigest, err := ComputeIntegrationBindingSetDigest(values)
	if err != nil {
		return err
	}
	if storedInstallationID != operation.AppInstallationID ||
		count != uint64(len(values)) ||
		storedDigest != currentDigest {
		return fmt.Errorf(
			"%w: operation IntegrationBinding set disagrees with its committed selection",
			ErrConflict,
		)
	}
	return nil
}