package runtimeapi

import (
	"context"
	"errors"

	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	recordsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	"google.golang.org/grpc"
)

// CatalogConsumerSelection is a deployment assertion for a closed
// workload-consumer capability. Catalog's active ConsumerGrant and immutable
// view remain the authorization and revision owners at request time.
type CatalogConsumerSelection struct {
	OrgID             string
	AppInstallationID string
	WorkloadID        string
	CapabilityID      string
}

type catalogConsumerRequestScopeContextKey struct{}

// CatalogConsumerAdmission binds the host-authenticated workload to one exact
// AppInstallation before a Catalog view may be disclosed. It owns no grant,
// view, cache, or revision selector.
type CatalogConsumerAdmission struct {
	byCapability map[string]CatalogConsumerSelection
}

func NewCatalogConsumerAdmission(
	selections []CatalogConsumerSelection,
) (*CatalogConsumerAdmission, error) {
	if len(selections) == 0 {
		return nil, errors.New(
			"runtime Catalog consumer admission: at least one selection is required",
		)
	}
	admission := &CatalogConsumerAdmission{
		byCapability: make(map[string]CatalogConsumerSelection, len(selections)),
	}
	for _, selection := range selections {
		if selection.OrgID == "" || selection.AppInstallationID == "" ||
			selection.WorkloadID == "" ||
			!catalogConsumerCapability(selection.CapabilityID) {
			return nil, errors.New(
				"runtime Catalog consumer admission: selection is incomplete",
			)
		}
		if _, duplicate := admission.byCapability[selection.CapabilityID]; duplicate {
			return nil, errors.New(
				"runtime Catalog consumer admission: capability is selected more than once",
			)
		}
		admission.byCapability[selection.CapabilityID] = selection
	}
	return admission, nil
}

func (a *CatalogConsumerAdmission) Ready() bool {
	return a != nil && len(a.byCapability) != 0
}

func (a *CatalogConsumerAdmission) UnaryServerInterceptor() grpc.UnaryServerInterceptor {
	return func(
		ctx context.Context,
		request any,
		_ *grpc.UnaryServerInfo,
		handler grpc.UnaryHandler,
	) (any, error) {
		capability, ok := runtimegrpc.Capability(ctx)
		if !ok {
			return nil, runtimegrpc.StatusError(ctx, &recordsemantic.Failure{
				Category: recordsemantic.FailureInternal,
			})
		}
		if !catalogConsumerCapability(capability) {
			return handler(ctx, request)
		}
		// One Runtime endpoint may serve Catalog views through two admitted
		// owners: static consumer selections (for Integration) and immutable
		// ServingAdmission selections (for an L1 AppService). This interceptor
		// must bind only the capabilities it owns so the next interceptor can
		// establish the other reviewed scope.
		if !a.owns(capability) {
			return handler(ctx, request)
		}
		admitted, err := a.bind(ctx, capability)
		if err != nil {
			return nil, runtimegrpc.StatusError(ctx, err)
		}
		return handler(admitted, request)
	}
}

func (a *CatalogConsumerAdmission) StreamServerInterceptor() grpc.StreamServerInterceptor {
	return func(
		server any,
		stream grpc.ServerStream,
		_ *grpc.StreamServerInfo,
		handler grpc.StreamHandler,
	) error {
		if stream == nil {
			return runtimegrpc.StatusError(context.Background(), &recordsemantic.Failure{
				Category: recordsemantic.FailureInternal,
			})
		}
		capability, ok := runtimegrpc.Capability(stream.Context())
		if !ok {
			return runtimegrpc.StatusError(stream.Context(), &recordsemantic.Failure{
				Category: recordsemantic.FailureInternal,
			})
		}
		if !catalogConsumerCapability(capability) {
			return handler(server, stream)
		}
		if !a.owns(capability) {
			return handler(server, stream)
		}
		admitted, err := a.bind(stream.Context(), capability)
		if err != nil {
			return runtimegrpc.StatusError(stream.Context(), err)
		}
		return handler(
			server,
			&runtimeContextServerStream{ServerStream: stream, ctx: admitted},
		)
	}
}

func (a *CatalogConsumerAdmission) owns(capability string) bool {
	if a == nil {
		return false
	}
	_, owned := a.byCapability[capability]
	return owned
}

// flowAppMetadataScope binds an internal Flow metadata read to the same
// deployment-selected workload grant used by the public Catalog consumer RPC.
// End-user delegation capabilities deliberately do not contain Catalog view
// grants; treating them as workload disclosure authority would either reject a
// valid split-process call or leak a workload-only capability into the user
// session. V3 is selected when configured, otherwise the N-1 V2 selection is
// used. A selected version never falls back after a read or validation error.
func (a *CatalogConsumerAdmission) flowAppMetadataScope(
	trusted TrustedCall,
) (CatalogConsumerScope, string, error) {
	if a == nil {
		return CatalogConsumerScope{}, "", &recordsemantic.Failure{
			Category: recordsemantic.FailureUnavailable,
		}
	}
	for _, capability := range []string{
		catalogsemantic.AppMetadataV3CapabilityName,
		catalogsemantic.AppMetadataV2CapabilityName,
	} {
		selection, found := a.byCapability[capability]
		if !found {
			continue
		}
		if trusted.orgID != selection.OrgID ||
			trusted.appInstallationID != selection.AppInstallationID ||
			trusted.callerWorkloadID != selection.WorkloadID {
			return CatalogConsumerScope{}, "", &recordsemantic.Failure{
				Category: recordsemantic.FailureForbidden,
			}
		}
		return CatalogConsumerScope{
			OrgID: selection.OrgID, AppInstallationID: selection.AppInstallationID,
			WorkloadID: selection.WorkloadID, Capabilities: []string{capability},
		}, capability, nil
	}
	return CatalogConsumerScope{}, "", &recordsemantic.Failure{
		Category: recordsemantic.FailureUnavailable,
	}
}

func (a *CatalogConsumerAdmission) bind(
	ctx context.Context,
	capability string,
) (context.Context, error) {
	if a == nil {
		return nil, &recordsemantic.Failure{
			Category: recordsemantic.FailureUnavailable,
		}
	}
	selection, found := a.byCapability[capability]
	trusted, trustedOK := trustedWorkloadCallFromContext(ctx)
	if !found || !trustedOK {
		return nil, &recordsemantic.Failure{
			Category: recordsemantic.FailureUnavailable,
		}
	}
	if trusted.orgID != selection.OrgID ||
		trusted.callerWorkloadID != selection.WorkloadID ||
		trusted.capability != selection.CapabilityID {
		return nil, &recordsemantic.Failure{
			Category: recordsemantic.FailureForbidden,
		}
	}
	scope := CatalogConsumerScope{
		OrgID: selection.OrgID, AppInstallationID: selection.AppInstallationID,
		WorkloadID: selection.WorkloadID, Capabilities: []string{capability},
	}
	return context.WithValue(
		ctx,
		catalogConsumerRequestScopeContextKey{},
		scope,
	), nil
}

func catalogConsumerScopeFromAdmission(
	ctx context.Context,
	capability string,
) (CatalogConsumerScope, bool) {
	if ctx == nil {
		return CatalogConsumerScope{}, false
	}
	scope, ok := ctx.Value(
		catalogConsumerRequestScopeContextKey{},
	).(CatalogConsumerScope)
	return scope, ok && scope.OrgID != "" && scope.AppInstallationID != "" &&
		scope.WorkloadID != "" && len(scope.Capabilities) == 1 &&
		scope.Capabilities[0] == capability
}

func catalogConsumerCapability(capability string) bool {
	switch capability {
	case catalogsemantic.AppMetadataCapabilityName,
		catalogsemantic.AppMetadataV2CapabilityName,
		catalogsemantic.AppMetadataV3CapabilityName,
		catalogsemantic.IntegrationBindingCapabilityName:
		return true
	default:
		return false
	}
}