package runtimeapi

import (
	"context"
	"errors"

	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	recordsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/catalog/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
)

// CatalogConsumerScope contains only host- and serving-admitted facts. Request
// messages cannot supply or override any field.
type CatalogConsumerScope struct {
	OrgID             string
	AppInstallationID string
	WorkloadID        string
	Capabilities      []string
}

// AppMetadataViewEngine is the N-1 Catalog surface required by bounded Flow
// execution. V2 binds AppRelease ownership and effective L1+L2 metadata to one
// revision, so Runtime never joins independently fetched authorities. Keeping
// this narrower than the serving interface prevents Flow from depending on
// unrelated consumer projections.
type AppMetadataViewEngine interface {
	GetAppMetadataViewV2(context.Context, CatalogConsumerScope, catalogsemantic.GetRequest) (catalogsemantic.AppMetadataViewV2, error)
}

// appMetadataViewV3Engine is detected by Flow when the admitted installation
// carries the V3 capability. It remains separate from AppMetadataViewEngine so
// N-1 V2 engine implementations keep working; an admitted V3 call must never
// silently fall back to a different authority version.
type appMetadataViewV3Engine interface {
	GetAppMetadataViewV3(context.Context, CatalogConsumerScope, catalogsemantic.GetRequest) (catalogsemantic.AppMetadataViewV3, error)
}

// CatalogConsumerViewEngine is the narrow production adapter boundary to
// Catalog's durable, transaction-consistent consumer reader. Runtime never
// compiles or reconstructs these views. ServingState may retain one validated
// immutable view inside the exact atomically published serving set.
type CatalogConsumerViewEngine interface {
	AppMetadataViewEngine
	GetAppMetadataView(context.Context, CatalogConsumerScope, catalogsemantic.GetRequest) (catalogsemantic.AppMetadataView, error)
	NextAppMetadataView(context.Context, CatalogConsumerScope, catalogsemantic.WatchRequest) (catalogsemantic.AppMetadataView, error)
	NextAppMetadataViewV2(context.Context, CatalogConsumerScope, catalogsemantic.WatchRequest) (catalogsemantic.AppMetadataViewV2, error)
	GetAppMetadataViewV3(context.Context, CatalogConsumerScope, catalogsemantic.GetRequest) (catalogsemantic.AppMetadataViewV3, error)
	NextAppMetadataViewV3(context.Context, CatalogConsumerScope, catalogsemantic.WatchRequest) (catalogsemantic.AppMetadataViewV3, error)
	GetIntegrationBindingView(context.Context, CatalogConsumerScope, catalogsemantic.GetRequest) (catalogsemantic.IntegrationBindingView, error)
	NextIntegrationBindingView(context.Context, CatalogConsumerScope, catalogsemantic.WatchRequest) (catalogsemantic.IntegrationBindingView, error)
}

type CatalogConsumerViewServer struct {
	catalogv1.UnimplementedConsumerViewServiceServer
	catalogv1.UnimplementedAppMetadataViewV3ServiceServer
	engine CatalogConsumerViewEngine
}

func NewCatalogConsumerViewServer(engine CatalogConsumerViewEngine) (*CatalogConsumerViewServer, error) {
	if engine == nil {
		return nil, errors.New("runtime Catalog consumer views: engine is required")
	}
	return &CatalogConsumerViewServer{engine: engine}, nil
}

func (s *CatalogConsumerViewServer) GetAppMetadataView(
	ctx context.Context,
	wireRequest *catalogv1.GetAppMetadataViewRequest,
) (*catalogv1.GetAppMetadataViewResponse, error) {
	request, err := mapping.FromWireCatalogGetRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataCapabilityName)
	}
	scope, err := catalogConsumerScopeFromContext(ctx, catalogsemantic.AppMetadataCapabilityName)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataCapabilityName)
	}
	view, err := s.engine.GetAppMetadataView(ctx, scope, request)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(
			classifyPlatformEngineError(
				ctx, err, recordsemantic.InvalidComponentResponse,
			),
			catalogsemantic.AppMetadataCapabilityName,
		)
	}
	if err := validateAppMetadataViewScope(scope, request.MinCatalogRevision, view); err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataCapabilityName)
	}
	response, err := mapping.ToWireGetAppMetadataViewResponse(view)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataCapabilityName)
	}
	return response, nil
}

func (s *CatalogConsumerViewServer) WatchAppMetadataView(
	wireRequest *catalogv1.WatchAppMetadataViewRequest,
	stream catalogv1.ConsumerViewService_WatchAppMetadataViewServer,
) error {
	if stream == nil {
		return mapping.ToPlatformStatusError(
			&recordsemantic.Failure{Category: recordsemantic.FailureInternal},
			catalogsemantic.AppMetadataCapabilityName,
		)
	}
	ctx := stream.Context()
	request, err := mapping.FromWireAppMetadataWatchRequest(ctx, wireRequest)
	if err != nil {
		return mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataCapabilityName)
	}
	scope, err := catalogConsumerScopeFromContext(ctx, catalogsemantic.AppMetadataCapabilityName)
	if err != nil {
		return mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataCapabilityName)
	}
	state := catalogsemantic.NewAppMetadataWatchState(request)
	for {
		view, nextErr := s.engine.NextAppMetadataView(ctx, scope, request)
		if nextErr != nil {
			return mapping.ToPlatformStatusError(
				classifyPlatformEngineError(
					ctx, nextErr, recordsemantic.InvalidComponentResponse,
				),
				catalogsemantic.AppMetadataCapabilityName,
			)
		}
		if scopeErr := validateAppMetadataViewScope(scope, request.AfterCatalogRevision, view); scopeErr != nil {
			return mapping.ToPlatformStatusError(scopeErr, catalogsemantic.AppMetadataCapabilityName)
		}
		if observeErr := state.Observe(view); observeErr != nil {
			return mapping.ToPlatformStatusError(observeErr, catalogsemantic.AppMetadataCapabilityName)
		}
		response, mapErr := mapping.ToWireWatchAppMetadataViewResponse(view)
		if mapErr != nil {
			return mapping.ToPlatformStatusError(mapErr, catalogsemantic.AppMetadataCapabilityName)
		}
		if sendErr := stream.Send(response); sendErr != nil {
			return sendErr
		}
		revision := view.Lineage.CatalogRevision
		request.AfterCatalogRevision = &revision
	}
}

func (s *CatalogConsumerViewServer) GetAppMetadataViewV2(
	ctx context.Context,
	wireRequest *catalogv1.GetAppMetadataViewV2Request,
) (*catalogv1.GetAppMetadataViewV2Response, error) {
	request, err := mapping.FromWireCatalogGetV2Request(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(
			err,
			catalogsemantic.AppMetadataV2CapabilityName,
		)
	}
	scope, err := catalogConsumerScopeFromContext(
		ctx,
		catalogsemantic.AppMetadataV2CapabilityName,
	)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(
			err,
			catalogsemantic.AppMetadataV2CapabilityName,
		)
	}
	view, err := s.engine.GetAppMetadataViewV2(ctx, scope, request)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(
			classifyPlatformEngineError(
				ctx,
				err,
				recordsemantic.InvalidComponentResponse,
			),
			catalogsemantic.AppMetadataV2CapabilityName,
		)
	}
	if err := validateAppMetadataViewV2Scope(
		scope,
		request.MinCatalogRevision,
		view,
	); err != nil {
		return nil, mapping.ToPlatformStatusError(
			err,
			catalogsemantic.AppMetadataV2CapabilityName,
		)
	}
	response, err := mapping.ToWireGetAppMetadataViewV2Response(view)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(
			err,
			catalogsemantic.AppMetadataV2CapabilityName,
		)
	}
	return response, nil
}

func (s *CatalogConsumerViewServer) WatchAppMetadataViewV2(
	wireRequest *catalogv1.WatchAppMetadataViewV2Request,
	stream catalogv1.ConsumerViewService_WatchAppMetadataViewV2Server,
) error {
	if stream == nil {
		return mapping.ToPlatformStatusError(
			&recordsemantic.Failure{
				Category: recordsemantic.FailureInternal,
			},
			catalogsemantic.AppMetadataV2CapabilityName,
		)
	}
	ctx := stream.Context()
	request, err := mapping.FromWireAppMetadataV2WatchRequest(
		ctx,
		wireRequest,
	)
	if err != nil {
		return mapping.ToPlatformStatusError(
			err,
			catalogsemantic.AppMetadataV2CapabilityName,
		)
	}
	scope, err := catalogConsumerScopeFromContext(
		ctx,
		catalogsemantic.AppMetadataV2CapabilityName,
	)
	if err != nil {
		return mapping.ToPlatformStatusError(
			err,
			catalogsemantic.AppMetadataV2CapabilityName,
		)
	}
	state := catalogsemantic.NewAppMetadataV2WatchState(request)
	for {
		view, nextErr := s.engine.NextAppMetadataViewV2(
			ctx,
			scope,
			request,
		)
		if nextErr != nil {
			return mapping.ToPlatformStatusError(
				classifyPlatformEngineError(
					ctx,
					nextErr,
					recordsemantic.InvalidComponentResponse,
				),
				catalogsemantic.AppMetadataV2CapabilityName,
			)
		}
		if scopeErr := validateAppMetadataViewV2Scope(
			scope,
			request.AfterCatalogRevision,
			view,
		); scopeErr != nil {
			return mapping.ToPlatformStatusError(
				scopeErr,
				catalogsemantic.AppMetadataV2CapabilityName,
			)
		}
		if observeErr := state.Observe(view); observeErr != nil {
			return mapping.ToPlatformStatusError(
				observeErr,
				catalogsemantic.AppMetadataV2CapabilityName,
			)
		}
		response, mapErr :=
			mapping.ToWireWatchAppMetadataViewV2Response(view)
		if mapErr != nil {
			return mapping.ToPlatformStatusError(
				mapErr,
				catalogsemantic.AppMetadataV2CapabilityName,
			)
		}
		if sendErr := stream.Send(response); sendErr != nil {
			return sendErr
		}
		revision := view.Lineage.CatalogRevision
		request.AfterCatalogRevision = &revision
	}
}

func (s *CatalogConsumerViewServer) GetAppMetadataViewV3(
	ctx context.Context,
	wireRequest *catalogv1.GetAppMetadataViewV3Request,
) (*catalogv1.GetAppMetadataViewV3Response, error) {
	request, err := mapping.FromWireCatalogGetV3Request(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataV3CapabilityName)
	}
	scope, err := catalogConsumerScopeFromContext(ctx, catalogsemantic.AppMetadataV3CapabilityName)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataV3CapabilityName)
	}
	view, err := s.engine.GetAppMetadataViewV3(ctx, scope, request)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(
			classifyPlatformEngineError(ctx, err, recordsemantic.InvalidComponentResponse),
			catalogsemantic.AppMetadataV3CapabilityName,
		)
	}
	if err := validateAppMetadataViewV3Scope(scope, request.MinCatalogRevision, view); err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataV3CapabilityName)
	}
	response, err := mapping.ToWireGetAppMetadataViewV3Response(view)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataV3CapabilityName)
	}
	return response, nil
}

func (s *CatalogConsumerViewServer) WatchAppMetadataViewV3(
	wireRequest *catalogv1.WatchAppMetadataViewV3Request,
	stream catalogv1.AppMetadataViewV3Service_WatchAppMetadataViewV3Server,
) error {
	if stream == nil {
		return mapping.ToPlatformStatusError(
			&recordsemantic.Failure{Category: recordsemantic.FailureInternal},
			catalogsemantic.AppMetadataV3CapabilityName,
		)
	}
	ctx := stream.Context()
	request, err := mapping.FromWireAppMetadataV3WatchRequest(ctx, wireRequest)
	if err != nil {
		return mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataV3CapabilityName)
	}
	scope, err := catalogConsumerScopeFromContext(ctx, catalogsemantic.AppMetadataV3CapabilityName)
	if err != nil {
		return mapping.ToPlatformStatusError(err, catalogsemantic.AppMetadataV3CapabilityName)
	}
	state := catalogsemantic.NewAppMetadataV3WatchState(request)
	for {
		view, nextErr := s.engine.NextAppMetadataViewV3(ctx, scope, request)
		if nextErr != nil {
			return mapping.ToPlatformStatusError(
				classifyPlatformEngineError(ctx, nextErr, recordsemantic.InvalidComponentResponse),
				catalogsemantic.AppMetadataV3CapabilityName,
			)
		}
		if scopeErr := validateAppMetadataViewV3Scope(scope, request.AfterCatalogRevision, view); scopeErr != nil {
			return mapping.ToPlatformStatusError(scopeErr, catalogsemantic.AppMetadataV3CapabilityName)
		}
		if observeErr := state.Observe(view); observeErr != nil {
			return mapping.ToPlatformStatusError(observeErr, catalogsemantic.AppMetadataV3CapabilityName)
		}
		response, mapErr := mapping.ToWireWatchAppMetadataViewV3Response(view)
		if mapErr != nil {
			return mapping.ToPlatformStatusError(mapErr, catalogsemantic.AppMetadataV3CapabilityName)
		}
		if sendErr := stream.Send(response); sendErr != nil {
			return sendErr
		}
		revision := view.Lineage.CatalogRevision
		request.AfterCatalogRevision = &revision
	}
}

func (s *CatalogConsumerViewServer) GetIntegrationBindingView(
	ctx context.Context,
	wireRequest *catalogv1.GetIntegrationBindingViewRequest,
) (*catalogv1.GetIntegrationBindingViewResponse, error) {
	request, err := mapping.FromWireIntegrationBindingGetRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.IntegrationBindingCapabilityName)
	}
	scope, err := catalogConsumerScopeFromContext(ctx, catalogsemantic.IntegrationBindingCapabilityName)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.IntegrationBindingCapabilityName)
	}
	view, err := s.engine.GetIntegrationBindingView(ctx, scope, request)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(
			classifyPlatformEngineError(
				ctx, err, recordsemantic.InvalidComponentResponse,
			),
			catalogsemantic.IntegrationBindingCapabilityName,
		)
	}
	if err := validateIntegrationBindingViewScope(scope, request.MinCatalogRevision, view); err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.IntegrationBindingCapabilityName)
	}
	response, err := mapping.ToWireGetIntegrationBindingViewResponse(view)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, catalogsemantic.IntegrationBindingCapabilityName)
	}
	return response, nil
}

func (s *CatalogConsumerViewServer) WatchIntegrationBindingView(
	wireRequest *catalogv1.WatchIntegrationBindingViewRequest,
	stream catalogv1.ConsumerViewService_WatchIntegrationBindingViewServer,
) error {
	if stream == nil {
		return mapping.ToPlatformStatusError(
			&recordsemantic.Failure{Category: recordsemantic.FailureInternal},
			catalogsemantic.IntegrationBindingCapabilityName,
		)
	}
	ctx := stream.Context()
	request, err := mapping.FromWireIntegrationBindingWatchRequest(ctx, wireRequest)
	if err != nil {
		return mapping.ToPlatformStatusError(err, catalogsemantic.IntegrationBindingCapabilityName)
	}
	scope, err := catalogConsumerScopeFromContext(ctx, catalogsemantic.IntegrationBindingCapabilityName)
	if err != nil {
		return mapping.ToPlatformStatusError(err, catalogsemantic.IntegrationBindingCapabilityName)
	}
	state := catalogsemantic.NewIntegrationBindingWatchState(request)
	for {
		view, nextErr := s.engine.NextIntegrationBindingView(ctx, scope, request)
		if nextErr != nil {
			return mapping.ToPlatformStatusError(
				classifyPlatformEngineError(
					ctx, nextErr, recordsemantic.InvalidComponentResponse,
				),
				catalogsemantic.IntegrationBindingCapabilityName,
			)
		}
		if scopeErr := validateIntegrationBindingViewScope(scope, request.AfterCatalogRevision, view); scopeErr != nil {
			return mapping.ToPlatformStatusError(scopeErr, catalogsemantic.IntegrationBindingCapabilityName)
		}
		if observeErr := state.Observe(view); observeErr != nil {
			return mapping.ToPlatformStatusError(observeErr, catalogsemantic.IntegrationBindingCapabilityName)
		}
		response, mapErr := mapping.ToWireWatchIntegrationBindingViewResponse(view)
		if mapErr != nil {
			return mapping.ToPlatformStatusError(mapErr, catalogsemantic.IntegrationBindingCapabilityName)
		}
		if sendErr := stream.Send(response); sendErr != nil {
			return sendErr
		}
		revision := view.Lineage.CatalogRevision
		request.AfterCatalogRevision = &revision
	}
}

func validateAppMetadataViewScope(
	scope CatalogConsumerScope,
	minimumRevision *uint64,
	view catalogsemantic.AppMetadataView,
) error {
	if view.OrgID != scope.OrgID || view.AppInstallationID != scope.AppInstallationID {
		return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
	}
	return requireCatalogRevision(minimumRevision, view.Lineage.CatalogRevision)
}

func validateAppMetadataViewV2Scope(
	scope CatalogConsumerScope,
	minimumRevision *uint64,
	view catalogsemantic.AppMetadataViewV2,
) error {
	if view.OrgID != scope.OrgID ||
		view.AppInstallationID != scope.AppInstallationID {
		return &recordsemantic.Failure{
			Category: recordsemantic.FailureInternal,
		}
	}
	return requireCatalogRevision(
		minimumRevision,
		view.Lineage.CatalogRevision,
	)
}

func validateAppMetadataViewV3Scope(
	scope CatalogConsumerScope,
	minimumRevision *uint64,
	view catalogsemantic.AppMetadataViewV3,
) error {
	if view.OrgID != scope.OrgID || view.AppInstallationID != scope.AppInstallationID {
		return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
	}
	return requireCatalogRevision(minimumRevision, view.Lineage.CatalogRevision)
}

func validateIntegrationBindingViewScope(
	scope CatalogConsumerScope,
	minimumRevision *uint64,
	view catalogsemantic.IntegrationBindingView,
) error {
	if view.OrgID != scope.OrgID || view.AppInstallationID != scope.AppInstallationID ||
		view.AuthorizedWorkloadID != scope.WorkloadID {
		return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
	}
	return requireCatalogRevision(minimumRevision, view.Lineage.CatalogRevision)
}

func catalogConsumerScopeFromContext(ctx context.Context, capability string) (CatalogConsumerScope, error) {
	if scope, ok := catalogConsumerScopeFromAdmission(ctx, capability); ok {
		return scope, nil
	}
	scope, ok := runtimeExecutionScopeFromContext(ctx)
	if !ok || scope.orgID == "" || scope.appInstallationID == "" ||
		scope.workloadID == "" || scope.capability != capability {
		return CatalogConsumerScope{}, &recordsemantic.Failure{Category: recordsemantic.FailureUnavailable}
	}
	return CatalogConsumerScope{
		OrgID: scope.orgID, AppInstallationID: scope.appInstallationID,
		WorkloadID: scope.workloadID, Capabilities: []string{capability},
	}, nil
}

var _ catalogv1.ConsumerViewServiceServer = (*CatalogConsumerViewServer)(nil)