package catalogapp

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

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

// IsConsumerCapability reports whether capabilityID names one of Catalog's
// closed consumer views. Workload syntax and operation authority are separate
// facts and cannot affect this classification.
func IsConsumerCapability(capabilityID string) bool {
	return catalogconsumerauthority.IsConsumerCapability(capabilityID)
}

func canonicalOperationConsumerGrants(
	provided []appauthority.ProvidedCapability,
	requiredCapabilities []string,
	handlerWorkloadID string,
	workloadHandlerAdmissions []WorkloadHandlerAdmission,
	explicit []ConsumerGrant,
	authorizedWorkloads []string,
) ([]ConsumerGrant, error) {
	hasHandler := slices.ContainsFunc(provided, func(value appauthority.ProvidedCapability) bool {
		return value.Kind == appauthority.CapabilityKindServiceHandler
	})
	if hasHandler {
		if err := catalogconsumerauthority.ValidateWorkloadID(handlerWorkloadID); err != nil {
			return nil, fmt.Errorf("handler workload ID: %w", err)
		}
		if !slices.Contains(authorizedWorkloads, handlerWorkloadID) {
			return nil, fmt.Errorf("handler workload is not authorized by the App operation")
		}
	} else if handlerWorkloadID != "" {
		return nil, fmt.Errorf("App operation without service handlers cannot declare a handler workload")
	}

	admittedWorkloads := make(map[string]string, len(workloadHandlerAdmissions))
	for _, admission := range workloadHandlerAdmissions {
		admittedWorkloads[admission.CapabilityID] = admission.AuthorizedWorkloadID
	}
	values := append([]ConsumerGrant(nil), explicit...)
	for _, capability := range provided {
		if capability.Kind != appauthority.CapabilityKindServiceHandler {
			continue
		}
		workloadID := handlerWorkloadID
		if admittedWorkloadID, admitted := admittedWorkloads[capability.CapabilityID]; admitted {
			workloadID = admittedWorkloadID
		}
		for _, dependency := range capability.RequiredRuntimeCapabilities {
			if !IsConsumerCapability(dependency) {
				continue
			}
			values = append(values, ConsumerGrant{
				AuthorizedWorkloadID: workloadID,
				CapabilityID:         dependency,
			})
		}
	}
	for index, value := range values {
		if !slices.Contains(requiredCapabilities, value.CapabilityID) {
			return nil, fmt.Errorf(
				"consumer grant[%d] capability is not required by the verified App release",
				index,
			)
		}
	}
	return canonicalConsumerGrants(values, authorizedWorkloads)
}

func validateHandlerConsumerGrantAuthority(
	requiredCapabilities []string,
	workloadID string,
	grants []ConsumerGrant,
) error {
	for _, capabilityID := range requiredCapabilities {
		if IsConsumerCapability(capabilityID) &&
			!hasConsumerGrant(grants, workloadID, capabilityID) {
			return fmt.Errorf(
				"%w: handler workload has no %q Catalog consumer grant",
				ErrConflict, capabilityID,
			)
		}
	}
	return nil
}

func canonicalConsumerGrants(
	values []ConsumerGrant,
	authorizedWorkloads []string,
) ([]ConsumerGrant, error) {
	result, err := canonicalConsumerGrantShape(values)
	if err != nil {
		return nil, err
	}
	for index, value := range result {
		if !slices.Contains(authorizedWorkloads, value.AuthorizedWorkloadID) {
			return nil, fmt.Errorf(
				"consumer grant[%d] workload is not authorized by the App operation",
				index,
			)
		}
	}
	return result, nil
}

func canonicalConsumerGrantShape(values []ConsumerGrant) ([]ConsumerGrant, error) {
	result := append([]ConsumerGrant(nil), values...)
	for index, value := range result {
		if err := ValidateConsumerGrant(value); err != nil {
			return nil, fmt.Errorf("consumer grant[%d]: %w", index, err)
		}
	}
	if _, err := effectiveMetadataCapabilityForConsumerGrants(result); err != nil {
		return nil, err
	}
	sort.Slice(result, func(left, right int) bool {
		return consumerGrantKey(result[left]) < consumerGrantKey(result[right])
	})
	result = slices.CompactFunc(result, func(left, right ConsumerGrant) bool {
		return consumerGrantKey(left) == consumerGrantKey(right)
	})
	return result, nil
}

func effectiveMetadataCapabilityForConsumerGrants(
	values []ConsumerGrant,
) (string, error) {
	var hasV2, hasV3 bool
	for _, value := range values {
		switch value.CapabilityID {
		case consumerv1.AppMetadataV2CapabilityName:
			hasV2 = true
		case consumerv1.AppMetadataV3CapabilityName:
			hasV3 = true
		}
	}
	if hasV2 && hasV3 {
		return "", fmt.Errorf(
			"consumer grants cannot mix App metadata V2 and V3 in one operation",
		)
	}
	if hasV3 {
		return consumerv1.AppMetadataV3CapabilityName, nil
	}
	if hasV2 {
		return consumerv1.AppMetadataV2CapabilityName, nil
	}
	return "", nil
}

// ValidateConsumerGrant validates one closed Catalog disclosure grant without
// assigning workload authority. Callers must separately prove that the
// workload belongs to the operation or active installation.
func ValidateConsumerGrant(value ConsumerGrant) error {
	return catalogconsumerauthority.ValidateConsumerGrant(value)
}

func consumerGrantKey(value ConsumerGrant) string {
	return value.AuthorizedWorkloadID + "\x00" + value.CapabilityID
}

func hasConsumerGrant(values []ConsumerGrant, workloadID, capability string) bool {
	key := consumerGrantKey(ConsumerGrant{
		AuthorizedWorkloadID: workloadID,
		CapabilityID:         capability,
	})
	index, found := slices.BinarySearchFunc(
		values,
		key,
		func(value ConsumerGrant, target string) int {
			switch current := consumerGrantKey(value); {
			case current < target:
				return -1
			case current > target:
				return 1
			default:
				return 0
			}
		},
	)
	return found && index < len(values)
}

// ComputeConsumerGrantSetDigest returns the durable commitment over one
// canonical operation-bound Catalog consumer grant set.
func ComputeConsumerGrantSetDigest(values []ConsumerGrant) (string, error) {
	return catalogconsumerauthority.ComputeConsumerGrantSetDigest(values)
}

func stageConsumerGrantsTx(
	ctx context.Context,
	tx *sql.Tx,
	operationID string,
	installationID string,
	orgID string,
	values []ConsumerGrant,
) error {
	for _, value := range values {
		grantID := hashKey(
			"catalog-consumer-grant", operationID,
			value.AuthorizedWorkloadID, value.CapabilityID,
		)
		_, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogConsumerGrant
(ConsumerGrantId, OrgId, AppInstallationId, OperationId,
AuthorizedWorkloadId, CapabilityId,
CreatedDate)
VALUES (?, ?, ?, ?, ?, ?,
CURRENT_TIMESTAMP(6))`,
			grantID, orgID, installationID, operationID,
			value.AuthorizedWorkloadID, value.CapabilityID,
		)
		if isDuplicate(err) {
			return ErrConflict
		}
		if err != nil {
			return err
		}
	}
	setDigest, err := ComputeConsumerGrantSetDigest(values)
	if err != nil {
		return err
	}
	_, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogConsumerGrantSet
(OperationId, AppInstallationId, GrantSetDigest, GrantCount, CreatedDate)
VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
		operationID, installationID, setDigest, epochText(uint64(len(values))),
	)
	if isDuplicate(err) {
		return ErrConflict
	}
	return err
}

func ensureConsumerGrantSetForOperationRetryTx(
	ctx context.Context,
	tx *sql.Tx,
	operation Operation,
	values []ConsumerGrant,
) error {
	expectedDigest, err := ComputeConsumerGrantSetDigest(values)
	if err != nil {
		return err
	}
	var (
		installationID string
		storedDigest   string
		rawCount       []byte
	)
	err = tx.QueryRowContext(ctx, `SELECT
AppInstallationId, GrantSetDigest, GrantCount
FROM _System_CatalogConsumerGrantSet
WHERE OperationId = ?
FOR UPDATE`, operation.OperationID).Scan(
		&installationID, &storedDigest, &rawCount,
	)
	if errors.Is(err, sql.ErrNoRows) {
		if len(values) != 0 {
			return fmt.Errorf(
				"%w: non-empty consumer grant selection has no durable commitment",
				ErrConflict,
			)
		}
		_, err = tx.ExecContext(ctx, `INSERT INTO _System_CatalogConsumerGrantSet
(OperationId, AppInstallationId, GrantSetDigest, GrantCount, 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: consumer grant commitment disagrees with exact operation retry",
			ErrConflict,
		)
	}
	return nil
}

// ReadOperationConsumerGrants returns the exact immutable grant set committed
// by Catalog admission. Reconcile consumes this result instead of reconstructing
// handler disclosure from deployment input.
func (a *Authority) ReadOperationConsumerGrants(
	ctx context.Context,
	operationID string,
) ([]ConsumerGrant, error) {
	tx, err := a.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer rollback(tx)
	operation, err := a.readOperation(ctx, tx, operationID, false)
	if err != nil {
		return nil, err
	}
	values, err := readOperationConsumerGrantsTx(ctx, tx, operation)
	if err != nil {
		return nil, err
	}
	canonical, err := canonicalConsumerGrantShape(values)
	if err != nil || !slices.Equal(values, canonical) {
		return nil, fmt.Errorf("%w: operation consumer grant set is not canonical", ErrCorrupt)
	}
	// Empty selections were not committed by the pre-consumer-grant request
	// grammar. Reuse the existing exact-retry rule: only a truly empty legacy
	// selection may acquire the canonical empty commitment; a non-empty set
	// without its commitment still fails closed.
	if err := ensureConsumerGrantSetForOperationRetryTx(ctx, tx, operation, values); err != nil {
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return values, nil
}

func readOperationConsumerGrantsTx(
	ctx context.Context,
	tx *sql.Tx,
	operation Operation,
) ([]ConsumerGrant, error) {
	rows, err := tx.QueryContext(ctx, `SELECT
AuthorizedWorkloadId, CapabilityId, OrgId, AppInstallationId
FROM _System_CatalogConsumerGrant
WHERE OperationId = ?
ORDER BY BINARY AuthorizedWorkloadId, BINARY CapabilityId
FOR UPDATE`, operation.OperationID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var result []ConsumerGrant
	for rows.Next() {
		var (
			value             ConsumerGrant
			orgID             string
			appInstallationID string
		)
		if err := rows.Scan(
			&value.AuthorizedWorkloadID,
			&value.CapabilityID,
			&orgID,
			&appInstallationID,
		); err != nil {
			return nil, err
		}
		if orgID != operation.OrgID ||
			appInstallationID != operation.AppInstallationID {
			return nil, fmt.Errorf(
				"%w: consumer grant disagrees with its operation",
				ErrConflict,
			)
		}
		result = append(result, value)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return result, nil
}

func verifyOperationConsumerGrantsTx(
	ctx context.Context,
	tx *sql.Tx,
	operation Operation,
) error {
	values, err := readOperationConsumerGrantsTx(ctx, tx, operation)
	if err != nil {
		return err
	}
	authorizedWorkloads, err := readOperationAuthorizedWorkloadsTx(
		ctx, tx, operation,
	)
	if err != nil {
		return err
	}
	canonical, err := canonicalConsumerGrants(
		values, authorizedWorkloads,
	)
	if err != nil || !slices.Equal(values, canonical) {
		return fmt.Errorf("%w: operation consumer grant set is invalid", ErrCorrupt)
	}
	return verifyOperationConsumerGrantCommitmentTx(ctx, tx, operation, values)
}

func verifyOperationConsumerGrantCommitmentTx(
	ctx context.Context,
	tx *sql.Tx,
	operation Operation,
	values []ConsumerGrant,
) error {
	var (
		storedInstallationID string
		storedDigest         string
		rawCount             []byte
	)
	err := tx.QueryRowContext(ctx, `SELECT
AppInstallationId, GrantSetDigest, GrantCount
FROM _System_CatalogConsumerGrantSet
WHERE OperationId = ?
FOR UPDATE`, operation.OperationID).Scan(
		&storedInstallationID, &storedDigest, &rawCount,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf("%w: operation has no consumer grant set commitment", ErrCorrupt)
	}
	if err != nil {
		return err
	}
	count, err := parseEpochAllowZero(rawCount)
	if err != nil {
		return err
	}
	currentDigest, err := ComputeConsumerGrantSetDigest(values)
	if err != nil {
		return err
	}
	if storedInstallationID != operation.AppInstallationID ||
		count != uint64(len(values)) ||
		storedDigest != currentDigest {
		return fmt.Errorf(
			"%w: operation consumer grant set disagrees with its committed selection",
			ErrConflict,
		)
	}
	return nil
}

func readOperationAuthorizedWorkloadsTx(
	ctx context.Context,
	tx *sql.Tx,
	operation Operation,
) ([]string, error) {
	var (
		orgID             string
		appID             string
		activeOperationID sql.NullString
		rawWorkloads      []byte
	)
	err := tx.QueryRowContext(ctx, `SELECT
OrgId, AppId, ActiveOperationId, CandidateAuthorizedWorkloads
FROM _System_CatalogAppInstallation
WHERE AppInstallationId = ?
FOR UPDATE`, operation.AppInstallationID).Scan(
		&orgID, &appID, &activeOperationID, &rawWorkloads,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, fmt.Errorf("%w: consumer grant installation is missing", ErrCorrupt)
	}
	if err != nil {
		return nil, err
	}
	if orgID != operation.OrgID ||
		appID != operation.AppID ||
		activeOperationID.String != operation.OperationID {
		return nil, fmt.Errorf(
			"%w: consumer grant operation is not the installation candidate",
			ErrConflict,
		)
	}
	if operation.Kind == OperationRemove {
		if len(rawWorkloads) != 0 {
			return nil, fmt.Errorf(
				"%w: remove operation has candidate workload authority",
				ErrCorrupt,
			)
		}
		return nil, nil
	}
	var workloads []string
	if err := json.Unmarshal(rawWorkloads, &workloads); err != nil {
		return nil, fmt.Errorf(
			"%w: decode candidate workload authority: %v",
			ErrCorrupt, err,
		)
	}
	canonical, err := canonicalStrings(
		"candidate authorized workloads", workloads, true,
	)
	if err != nil || !slices.Equal(workloads, canonical) {
		return nil, fmt.Errorf(
			"%w: candidate workload authority is not canonical",
			ErrCorrupt,
		)
	}
	return workloads, nil
}