package runtimeapi

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"sync/atomic"
	"time"

	"github.com/nexuscrm/backend/internal/managedruntime"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	approvalsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalv1"
	auditsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/auditlogv1"
	bulkingestsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	permissionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/effectivepermissionsv1"
	filessemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/filesv1"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	listviewsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/listviewv1"
	mailboxauthorizationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/mailboxauthorizationv1"
	permissionauthoringsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionauthoringv1"
	querysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
	accesssemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordaccessv1"
	attachmentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordattachmentsv1"
	defaultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recorddefaultsv1"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	recordviewsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordviewv1"
	recyclemutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recyclebinmutationv1"
	recyclesemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recyclebinv1"
	relatedsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/relatedcountsv1"
	reportsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/reportv1"
	searchsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/searchv1"
	"google.golang.org/grpc"
)

// flowAppMetadataPrepareAttemptCap bounds the one durable Catalog read used
// to build an immutable serving set. Public Catalog reads arrive with a gRPC
// deadline; startup and security-epoch reconciliation use a process context,
// so this owner must add the equivalent bounded attempt itself.
const flowAppMetadataPrepareAttemptCap = 30 * time.Second

// ServingInstallation is the Catalog-owned installation fact required to
// prepare one Runtime capability. Runtime does not author or infer any field.
type ServingInstallation struct {
	OrgID               string
	AppInstallationID   string
	ActiveReleaseDigest string
	CatalogRevision     uint64
}

// ServingHandlerBinding is the Catalog-owned active workload owner for one
// installed capability.
type ServingHandlerBinding struct {
	OrgID                      string
	AppInstallationID          string
	AppReleaseDigest           string
	CapabilityID               string
	WorkloadID                 string
	ServiceReleaseDigest       string
	OwnershipEpoch             uint64
	CompatibilityReceiptDigest string
}

// ServingCatalogSource returns one statement-consistent installation/binding
// tuple. Runtime must never assemble these activation-coupled facts through
// independent reads.
type ServingCatalogSource func(context.Context, string, string) (ServingInstallation, ServingHandlerBinding, bool, error)

// ServingSelection is deployment-owned selection, not mutable serving
// authority. The configured receipt attests the deployment, while Catalog
// selects its current metadata publication. Reconcile verifies both before
// the complete set can become ready.
type ServingSelection struct {
	OrgID             string
	AppInstallationID string
	// ExpectedWorkloadID is a deployment assertion, not a second workload
	// authority. Reconcile must match it to Catalog's active HandlerBinding
	// before this replica can become ready.
	ExpectedWorkloadID string
	// AppCapabilityID selects the Catalog-owned L1 HandlerBinding. It is
	// deliberately distinct from the L0 Runtime capability used by the data
	// request and metadata projection.
	AppCapabilityID string
	// RuntimeCapabilityID selects the delegation, RuntimeMetadataView, and
	// response receipt. Catalog must never infer it from the handler ID.
	RuntimeCapabilityID                string
	ExpectedServiceReleaseDigest       catalogsnapshot.Digest
	ExpectedOwnershipEpoch             uint64
	ExpectedCompatibilityReceiptDigest catalogsnapshot.Digest
	LocalCompatibility                 catalogsnapshot.Compatibility
	SecurityEpoch                      uint64
}

type ServingStateConfig struct {
	UserAPICallers              []string
	Consumers                   []ServingConsumerSelection
	ConsumerViews               CatalogConsumerViewEngine
	Store                       catalogsnapshot.Store
	Catalog                     ServingCatalogSource
	AppMetadata                 AppMetadataViewEngine
	AssignmentPolicyEligibility func(context.Context, *catalogsnapshot.RuntimeMetadataView) (managedruntime.AssignmentPolicyEligibilityAmbientStager, error)
}

type preparedServingState struct {
	deploymentSnapshotDigest    catalogsnapshot.Digest
	selection                   ServingSelection
	binding                     ServingHandlerBinding
	sourceSnapshotDigest        catalogsnapshot.Digest
	view                        *catalogsnapshot.RuntimeMetadataView
	receipts                    semantic.Receipts
	assignmentPolicyEligibility managedruntime.AssignmentPolicyEligibilityAmbientStager
}

type preparedServingSet struct {
	commandConsumer    *preparedCommandConsumer
	outboundConsumer   *preparedOutboundConsumer
	pageLayoutConsumer *preparedPageLayoutConsumer
	byCapability       map[string]*preparedServingState
	flowAppMetadata    flowAppMetadataView
}

type runtimeEffectivePermissionMetadataViewContextKey struct{}

func effectivePermissionMetadataViewFromContext(ctx context.Context) (*catalogsnapshot.RuntimeMetadataView, bool) {
	view, ok := ctx.Value(runtimeEffectivePermissionMetadataViewContextKey{}).(*catalogsnapshot.RuntimeMetadataView)
	return view, ok && view != nil
}

// ServingState owns the one prepared immutable capability view. Reconcile may
// consult Catalog and the artifact store; request admission performs only an
// atomic read of the already-verified result.
type ServingState struct {
	userAPICallers                     []string
	consumers                          []ServingConsumerSelection
	consumerViews                      CatalogConsumerViewEngine
	store                              catalogsnapshot.Store
	catalog                            ServingCatalogSource
	appMetadata                        AppMetadataViewEngine
	prepareAssignmentPolicyEligibility func(context.Context, *catalogsnapshot.RuntimeMetadataView) (managedruntime.AssignmentPolicyEligibilityAmbientStager, error)
	prepared                           atomic.Pointer[preparedServingSet]
}

func NewServingState(config ServingStateConfig) (*ServingState, error) {
	if config.Store == nil || config.Catalog == nil {
		return nil, errors.New("runtime serving state: store and Catalog serving authority are required")
	}
	if err := validateServingConsumers(config.Consumers); err != nil {
		return nil, err
	}
	if err := validateUserAPICallers(config.UserAPICallers); err != nil {
		return nil, err
	}
	return &ServingState{
		userAPICallers: slices.Clone(config.UserAPICallers),
		consumers:      append([]ServingConsumerSelection(nil), config.Consumers...), consumerViews: config.ConsumerViews,
		store: config.Store, catalog: config.Catalog, appMetadata: config.AppMetadata,
		prepareAssignmentPolicyEligibility: config.AssignmentPolicyEligibility,
	}, nil
}

// Reconcile prepares and atomically publishes one exact serving view. Any
// failure removes readiness for subsequent requests; requests that already
// captured the previous pointer remain revision-stable until completion.
func (s *ServingState) Reconcile(ctx context.Context, selection ServingSelection) error {
	return s.ReconcileAll(ctx, []ServingSelection{selection})
}

// ReconcileAll prepares a complete immutable capability set and publishes it
// with one atomic pointer swap. A request can therefore never observe a mixed
// old/new set or borrow another capability's metadata view.
func (s *ServingState) ReconcileAll(ctx context.Context, selections []ServingSelection) error {
	if s == nil {
		return errors.New("runtime serving state: state is unavailable")
	}
	if len(selections) == 0 {
		s.prepared.Store(nil)
		return errors.New("runtime serving state: at least one serving selection is required")
	}
	previous := s.prepared.Load()
	artifacts := newServingArtifactSet(s.store)
	set := &preparedServingSet{byCapability: make(map[string]*preparedServingState, len(selections))}
	for _, selection := range selections {
		prepared, err := s.prepare(ctx, selection, artifacts)
		if err != nil {
			s.prepared.Store(nil)
			return err
		}
		if _, duplicate := set.byCapability[selection.RuntimeCapabilityID]; duplicate {
			s.prepared.Store(nil)
			return errors.New("runtime serving state: Runtime capability is selected more than once")
		}
		set.byCapability[selection.RuntimeCapabilityID] = prepared
	}
	first := set.byCapability[selections[0].RuntimeCapabilityID]
	for _, prepared := range set.byCapability {
		if prepared.selection.OrgID != first.selection.OrgID ||
			prepared.selection.AppInstallationID != first.selection.AppInstallationID ||
			prepared.selection.SecurityEpoch != first.selection.SecurityEpoch ||
			prepared.binding.AppReleaseDigest != first.binding.AppReleaseDigest ||
			prepared.binding.ServiceReleaseDigest != first.binding.ServiceReleaseDigest ||
			prepared.deploymentSnapshotDigest != first.deploymentSnapshotDigest ||
			prepared.sourceSnapshotDigest != first.sourceSnapshotDigest ||
			prepared.receipts.CatalogRevision != first.receipts.CatalogRevision ||
			prepared.receipts.SchemaGeneration != first.receipts.SchemaGeneration ||
			prepared.receipts.EngineGeneration != first.receipts.EngineGeneration {
			s.prepared.Store(nil)
			return errors.New("runtime serving state: capability set does not share one installation, App/service release, Catalog snapshot lineage, and security epoch")
		}
	}
	capabilities := make([]string, 0, len(set.byCapability))
	for capability := range set.byCapability {
		capabilities = append(capabilities, capability)
	}
	if err := ValidateServingCapabilitySet(capabilities); err != nil {
		s.prepared.Store(nil)
		return fmt.Errorf("runtime serving state: %w", err)
	}
	if prepared := set.byCapability[mutationsemantic.CapabilityName]; prepared != nil {
		var previousPrepared *preparedServingState
		if previous != nil {
			previousPrepared = previous.byCapability[mutationsemantic.CapabilityName]
		}
		if sameAssignmentPolicyEligibilityServingView(previousPrepared, prepared) {
			prepared.assignmentPolicyEligibility = previousPrepared.assignmentPolicyEligibility
		} else if s.prepareAssignmentPolicyEligibility == nil {
			if runtimeViewDeclaresAssignmentPolicyEligibility(prepared.view) {
				s.prepared.Store(nil)
				return fmt.Errorf("runtime serving state: assignment-policy lifecycle preparation is unavailable for %s", mutationsemantic.CapabilityName)
			}
		} else {
			stager, err := s.prepareAssignmentPolicyEligibility(ctx, prepared.view)
			if err != nil {
				s.prepared.Store(nil)
				return fmt.Errorf("runtime serving state: prepare assignment-policy lifecycle binding for %s: %w", mutationsemantic.CapabilityName, err)
			}
			prepared.assignmentPolicyEligibility = stager
		}
	}
	if _, flowReady := set.byCapability[flowsemantic.CapabilityName]; flowReady {
		prepared, err := s.prepareFlowAppMetadata(ctx, set)
		if err != nil {
			s.prepared.Store(nil)
			return err
		}
		set.flowAppMetadata = prepared
	}
	if err := s.prepareConsumers(ctx, set, first, artifacts); err != nil {
		s.prepared.Store(nil)
		return err
	}
	s.prepared.Store(set)
	return nil
}

func sameAssignmentPolicyEligibilityServingView(left, right *preparedServingState) bool {
	if left == nil || right == nil {
		return false
	}
	return left.selection.OrgID == right.selection.OrgID &&
		left.selection.AppInstallationID == right.selection.AppInstallationID &&
		left.selection.RuntimeCapabilityID == right.selection.RuntimeCapabilityID &&
		left.binding.AppReleaseDigest == right.binding.AppReleaseDigest &&
		left.binding.ServiceReleaseDigest == right.binding.ServiceReleaseDigest &&
		left.sourceSnapshotDigest == right.sourceSnapshotDigest &&
		left.receipts.CatalogRevision == right.receipts.CatalogRevision &&
		left.receipts.SchemaGeneration == right.receipts.SchemaGeneration &&
		left.receipts.EngineGeneration == right.receipts.EngineGeneration
}

func (s *ServingState) prepareFlowAppMetadata(
	ctx context.Context,
	set *preparedServingSet,
) (flowAppMetadataView, error) {
	if s == nil || s.appMetadata == nil || set == nil {
		return flowAppMetadataView{}, errors.New(
			"runtime serving state: Flow execution requires a prepared AppMetadata reader",
		)
	}
	flowPrepared := set.byCapability[flowsemantic.CapabilityName]
	if flowPrepared == nil {
		return flowAppMetadataView{}, errors.New(
			"runtime serving state: Flow capability is unavailable",
		)
	}
	var appPrepared *preparedServingState
	var capability string
	for _, candidate := range []string{
		catalogsemantic.AppMetadataV3CapabilityName,
		catalogsemantic.AppMetadataV2CapabilityName,
	} {
		if prepared := set.byCapability[candidate]; prepared != nil {
			appPrepared, capability = prepared, candidate
			break
		}
	}
	if appPrepared == nil {
		return flowAppMetadataView{}, errors.New(
			"runtime serving state: Flow execution requires AppMetadata V3 or V2 authority",
		)
	}
	scope := CatalogConsumerScope{
		OrgID: appPrepared.selection.OrgID, AppInstallationID: appPrepared.selection.AppInstallationID,
		WorkloadID: appPrepared.binding.WorkloadID, Capabilities: []string{capability},
	}
	minimum := flowPrepared.receipts.CatalogRevision
	request := catalogsemantic.GetRequest{MinCatalogRevision: &minimum}
	readCtx, cancel := context.WithTimeout(ctx, flowAppMetadataPrepareAttemptCap)
	defer cancel()
	var normalized flowAppMetadataView
	switch capability {
	case catalogsemantic.AppMetadataV3CapabilityName:
		engine, ok := s.appMetadata.(appMetadataViewV3Engine)
		if !ok {
			return normalized, errors.New(
				"runtime serving state: selected AppMetadata V3 reader is unavailable",
			)
		}
		view, err := engine.GetAppMetadataViewV3(readCtx, scope, request)
		if err != nil {
			return normalized, fmt.Errorf("runtime serving state: prepare AppMetadata V3: %w", err)
		}
		if catalogsemantic.ValidateAppMetadataViewV3(view) != nil {
			return normalized, errors.New("runtime serving state: AppMetadata V3 view is invalid")
		}
		if view.AppReleaseDigest != appPrepared.binding.AppReleaseDigest {
			return normalized, errors.New(
				"runtime serving state: AppMetadata V3 release does not match serving authority",
			)
		}
		normalized = normalizeFlowAppMetadataV3(view)
	case catalogsemantic.AppMetadataV2CapabilityName:
		view, err := s.appMetadata.GetAppMetadataViewV2(readCtx, scope, request)
		if err != nil {
			return normalized, fmt.Errorf("runtime serving state: prepare AppMetadata V2: %w", err)
		}
		if catalogsemantic.ValidateAppMetadataViewV2(view) != nil {
			return normalized, errors.New("runtime serving state: AppMetadata V2 view is invalid")
		}
		if view.AppReleaseDigest != appPrepared.binding.AppReleaseDigest {
			return normalized, errors.New(
				"runtime serving state: AppMetadata V2 release does not match serving authority",
			)
		}
		normalized = normalizeFlowAppMetadataV2(view)
	}
	flowScope := RuntimeExecutionScope{
		view: flowPrepared.view, orgID: flowPrepared.selection.OrgID,
		appInstallationID: flowPrepared.selection.AppInstallationID,
		workloadID:        flowPrepared.binding.WorkloadID, capability: flowsemantic.CapabilityName,
		receipts: flowPrepared.receipts,
	}
	if err := validateFlowAppMetadataAuthority(flowScope, flowPrepared.view, normalized); err != nil {
		return flowAppMetadataView{}, fmt.Errorf(
			"runtime serving state: prepared AppMetadata authority does not match Flow: %w", err,
		)
	}
	return normalized, nil
}

func (s *ServingState) Ready() bool {
	if s == nil {
		return false
	}
	prepared := s.prepared.Load()
	return prepared != nil && len(prepared.byCapability) > 0
}

func (s *ServingState) prepare(ctx context.Context, selection ServingSelection, artifacts *servingArtifactSet) (*preparedServingState, error) {
	if err := validateServingSelection(selection); err != nil {
		return nil, err
	}
	installation, binding, found, err := s.catalog(ctx, selection.AppInstallationID, selection.AppCapabilityID)
	if err != nil {
		return nil, fmt.Errorf("runtime serving state: read Catalog serving capability: %w", err)
	}
	if !found {
		return nil, errors.New("runtime serving state: selected serving capability is unavailable")
	}
	anchor, err := s.deploymentReceipt(ctx, selection, artifacts)
	if err != nil {
		return nil, err
	}
	if string(anchor.AppReleaseDigest()) != installation.ActiveReleaseDigest ||
		binding.OwnershipEpoch < anchor.OwnershipEpoch() ||
		installation.CatalogRevision < uint64(anchor.Lineage().CatalogRevision) ||
		selection.SecurityEpoch < anchor.SecurityEpoch() {
		return nil, errors.New("runtime serving state: publication is outside the admitted deployment")
	}
	if (binding.OwnershipEpoch == anchor.OwnershipEpoch() || installation.CatalogRevision == uint64(anchor.Lineage().CatalogRevision)) &&
		binding.CompatibilityReceiptDigest != string(anchor.Digest()) {
		return nil, errors.New("runtime serving state: publication reuses an immutable deployment identity")
	}
	// These values belong to the statement-consistent active Catalog tuple.
	// The input selection remains the immutable deployment attestation.
	selection.ExpectedOwnershipEpoch = binding.OwnershipEpoch
	selection.ExpectedCompatibilityReceiptDigest = catalogsnapshot.Digest(binding.CompatibilityReceiptDigest)
	if installation.OrgID != selection.OrgID || installation.AppInstallationID != selection.AppInstallationID ||
		binding.OrgID != selection.OrgID || binding.AppInstallationID != selection.AppInstallationID ||
		binding.CapabilityID != selection.AppCapabilityID || binding.AppReleaseDigest == "" ||
		binding.AppReleaseDigest != installation.ActiveReleaseDigest || binding.WorkloadID != selection.ExpectedWorkloadID ||
		binding.ServiceReleaseDigest != string(selection.ExpectedServiceReleaseDigest) ||
		binding.OwnershipEpoch != selection.ExpectedOwnershipEpoch ||
		binding.CompatibilityReceiptDigest != string(selection.ExpectedCompatibilityReceiptDigest) {
		return nil, errors.New("runtime serving state: Catalog ownership facts are incomplete or inconsistent")
	}
	receipt, err := artifacts.receipt(ctx, selection.ExpectedCompatibilityReceiptDigest)
	if err != nil {
		return nil, fmt.Errorf("runtime serving state: load compatibility receipt: %w", err)
	}
	if receipt.Digest() != selection.ExpectedCompatibilityReceiptDigest ||
		receipt.OrgID() != selection.OrgID || receipt.AppInstallationID() != selection.AppInstallationID ||
		string(receipt.AppReleaseDigest()) != installation.ActiveReleaseDigest ||
		receipt.AppCapabilityID() != selection.AppCapabilityID ||
		receipt.RuntimeCapabilityID() != selection.RuntimeCapabilityID ||
		receipt.WorkloadID() != binding.WorkloadID ||
		receipt.ServiceReleaseDigest() != selection.ExpectedServiceReleaseDigest ||
		receipt.OwnershipEpoch() != selection.ExpectedOwnershipEpoch ||
		receipt.SecurityEpoch() != selection.SecurityEpoch {
		return nil, errors.New("runtime serving state: compatibility receipt does not match Catalog ownership")
	}
	snapshot, err := artifacts.snapshot(ctx, receipt.CatalogSnapshotDigest())
	if err != nil {
		return nil, fmt.Errorf("runtime serving state: load Catalog snapshot: %w", err)
	}
	view, err := artifacts.view(ctx, receipt.RuntimeMetadataViewDigest())
	if err != nil {
		return nil, fmt.Errorf("runtime serving state: load Runtime metadata view: %w", err)
	}
	if err := artifacts.verifySource(view, snapshot); err != nil {
		return nil, fmt.Errorf("runtime serving state: verify Runtime metadata source: %w", err)
	}
	if err := view.VerifyCompatibility(selection.LocalCompatibility); err != nil {
		return nil, fmt.Errorf("runtime serving state: verify Runtime compatibility: %w", err)
	}
	if err := validateServingView(selection.RuntimeCapabilityID, view); err != nil {
		return nil, err
	}
	lineage := view.Lineage()
	receiptLineage := receipt.Lineage()
	if snapshot.Digest() != receipt.CatalogSnapshotDigest() || view.Digest() != receipt.RuntimeMetadataViewDigest() ||
		snapshot.OrgID() != selection.OrgID || view.OrgID() != selection.OrgID ||
		view.AppInstallationID() != selection.AppInstallationID || view.CapabilityID() != selection.RuntimeCapabilityID ||
		lineage.CatalogRevision == 0 || lineage.SchemaGeneration == 0 || lineage.EngineGeneration == 0 ||
		uint64(lineage.CatalogRevision) != installation.CatalogRevision || lineage != receiptLineage {
		return nil, errors.New("runtime serving state: artifact scope or lineage does not match Catalog authority")
	}
	return &preparedServingState{
		deploymentSnapshotDigest: anchor.CatalogSnapshotDigest(),
		selection:                selection, binding: binding, sourceSnapshotDigest: view.SourceSnapshotDigest(), view: view,
		receipts: semantic.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: selection.RuntimeCapabilityID,
			CatalogRevision: uint64(lineage.CatalogRevision), SchemaGeneration: uint64(lineage.SchemaGeneration),
			EngineGeneration: uint64(lineage.EngineGeneration), SecurityEpoch: selection.SecurityEpoch,
		},
	}, nil
}

// validateServingView closes the gap between artifact compatibility and
// serving compatibility. Historical views remain readable for audit and
// rollback, but a capability cannot become ready unless its projection carries
// the complete grammar required by the current Runtime implementation.
func validateServingView(capability string, view *catalogsnapshot.RuntimeMetadataView) error {
	expected := servingProjectionForCapability(capability)
	if view == nil || expected == "" || view.Profile() != expected {
		actual := catalogsnapshot.ProjectionProfile("")
		if view != nil {
			actual = view.Profile()
		}
		return fmt.Errorf(
			"runtime serving state: Runtime metadata view has the wrong projection profile: capability=%q expected=%q actual=%q",
			capability, expected, actual,
		)
	}
	if expected == catalogsnapshot.ProjectionRecordViewV1 && !view.SupportsRecordViewServing() {
		return errors.New("runtime serving state: Runtime metadata view is not eligible for record-view serving")
	}
	if expected == catalogsnapshot.ProjectionRecordAccessV1 && !view.SupportsRecordAccessServing() {
		return errors.New("runtime serving state: Runtime metadata view is not eligible for record-access serving")
	}
	if expected == catalogsnapshot.ProjectionEffectivePermissionsV1 && !view.SupportsEffectivePermissionServing() {
		return errors.New("runtime serving state: Runtime metadata view is not eligible for effective-permission serving")
	}
	return nil
}

type servingCapabilitySpec struct {
	projection            catalogsnapshot.ProjectionProfile
	requiredSiblings      []string
	bindsMutationEnvelope bool
}

// servingCapabilitySpecs is the sole owner of immutable ServingAdmission
// composition. A capability in this registry has exactly one projection and
// declares every sibling authority that its engine can consume. Production
// selection validates against this registry before opening the database.
var servingCapabilitySpecs = map[string]servingCapabilitySpec{
	accesssemantic.CapabilityName:     {projection: catalogsnapshot.ProjectionRecordAccessV1},
	permissionsemantic.CapabilityName: {projection: catalogsnapshot.ProjectionEffectivePermissionsV1},
	auditsemantic.CapabilityName:      {projection: catalogsnapshot.ProjectionEffectivePermissionsV1},
	querysemantic.CapabilityName:      {projection: catalogsnapshot.ProjectionRecordQueryV1},
	relatedsemantic.CapabilityName: {
		projection:       catalogsnapshot.ProjectionRecordQueryV1,
		requiredSiblings: []string{accesssemantic.CapabilityName},
	},
	recyclesemantic.CapabilityName: {projection: catalogsnapshot.ProjectionRecordQueryV1},
	mutationsemantic.CapabilityName: {
		projection:            catalogsnapshot.ProjectionRecordMutationV1,
		bindsMutationEnvelope: true,
	},
	recyclemutationsemantic.CapabilityName: {
		projection:            catalogsnapshot.ProjectionRecordMutationV1,
		requiredSiblings:      []string{mutationsemantic.CapabilityName},
		bindsMutationEnvelope: true,
	},
	listviewsemantic.CapabilityName:                  {projection: catalogsnapshot.ProjectionRecordMutationV1},
	bulkingestsemantic.CapabilityName:                {projection: catalogsnapshot.ProjectionRecordMutationV1},
	defaultsemantic.CapabilityName:                   {projection: catalogsnapshot.ProjectionRecordMutationV1},
	recordviewsemantic.CapabilityName:                {projection: catalogsnapshot.ProjectionRecordViewV1},
	attachmentsemantic.CapabilityName:                {projection: catalogsnapshot.ProjectionRecordViewV1},
	catalogsemantic.AppMetadataCapabilityName:        {projection: catalogsnapshot.ProjectionRecordQueryV1},
	catalogsemantic.AppMetadataV2CapabilityName:      {projection: catalogsnapshot.ProjectionRecordQueryV1},
	catalogsemantic.AppMetadataV3CapabilityName:      {projection: catalogsnapshot.ProjectionRecordQueryV1},
	catalogsemantic.IntegrationBindingCapabilityName: {projection: catalogsnapshot.ProjectionRecordQueryV1},
	searchsemantic.CapabilityName:                    {projection: catalogsnapshot.ProjectionRecordQueryV1},
	commandsemantic.CapabilityName:                   {projection: catalogsnapshot.ProjectionCommandAcceptanceV1},
	reportsemantic.CapabilityName:                    {projection: catalogsnapshot.ProjectionReportExecutionV1},
	filessemantic.CapabilityName:                     {projection: catalogsnapshot.ProjectionRecordMutationV1},
	flowsemantic.CapabilityName: {
		projection:            catalogsnapshot.ProjectionRecordMutationV1,
		requiredSiblings:      []string{mutationsemantic.CapabilityName},
		bindsMutationEnvelope: true,
	},
	approvalsemantic.ExecuteCapabilityName: {
		projection:            catalogsnapshot.ProjectionApprovalExecutionV1,
		requiredSiblings:      []string{mutationsemantic.CapabilityName},
		bindsMutationEnvelope: true,
	},
	approvalsemantic.ReadCapabilityName: {projection: catalogsnapshot.ProjectionApprovalReadV1},
	permissionauthoringsemantic.CapabilityName: {
		projection:            catalogsnapshot.ProjectionRecordMutationV1,
		requiredSiblings:      []string{mutationsemantic.CapabilityName},
		bindsMutationEnvelope: true,
	},
}

func servingCapabilitySpecFor(capability string) (servingCapabilitySpec, bool) {
	spec, ok := servingCapabilitySpecs[capability]
	return spec, ok && spec.projection != ""
}

// UsesServingAdmission exposes only the production-composition fact. The
// projection and dependency details remain private to their runtime owner.
func UsesServingAdmission(capability string) bool {
	_, ok := servingCapabilitySpecFor(capability)
	return ok
}

// ValidateServingCapabilitySet closes cross-capability authority before any
// serving view is published. Production assembly calls the same owner before
// opening the database, while reconciliation repeats it against the exact
// atomically prepared set.
func ValidateServingCapabilitySet(capabilities []string) error {
	selected := make(map[string]struct{}, len(capabilities))
	for _, capability := range capabilities {
		if _, ok := servingCapabilitySpecFor(capability); !ok {
			return fmt.Errorf("capability %q has no ServingAdmission specification", capability)
		}
		selected[capability] = struct{}{}
	}
	for capability := range selected {
		spec, _ := servingCapabilitySpecFor(capability)
		for _, required := range spec.requiredSiblings {
			if _, ok := selected[required]; !ok {
				return fmt.Errorf(
					"capability %q requires sibling capability %q in the same immutable set",
					capability,
					required,
				)
			}
		}
	}
	return nil
}

func servingProjectionForCapability(capability string) catalogsnapshot.ProjectionProfile {
	spec, ok := servingCapabilitySpecFor(capability)
	if !ok {
		return ""
	}
	return spec.projection
}

func validateServingSelection(selection ServingSelection) error {
	if selection.OrgID == "" || selection.AppInstallationID == "" || selection.ExpectedWorkloadID == "" || selection.AppCapabilityID == "" ||
		selection.RuntimeCapabilityID == "" || selection.AppCapabilityID == selection.RuntimeCapabilityID ||
		selection.ExpectedServiceReleaseDigest == "" || selection.ExpectedOwnershipEpoch == 0 ||
		selection.ExpectedCompatibilityReceiptDigest == "" ||
		selection.LocalCompatibility.SchemaGeneration == 0 || selection.LocalCompatibility.EngineGeneration == 0 ||
		selection.SecurityEpoch == 0 {
		return errors.New("runtime serving state: every deployment selector and generation is required")
	}
	return nil
}

type runtimeExecutionScopeContextKey struct{}
type runtimeMutationMetadataViewContextKey struct{}
type runtimeFlowAppMetadataContextKey struct{}
type runtimeRecordAccessMetadataViewContextKey struct{}
type runtimeRecordsMetadataViewOverrideContextKey struct{}

// RuntimeExecutionScope is private to the Runtime process. It deliberately
// does not enter protobuf, public semantic contracts, contract.Deps, or an
// engine-wide interface.
type RuntimeExecutionScope struct {
	view              *catalogsnapshot.RuntimeMetadataView
	orgID             string
	appInstallationID string
	workloadID        string
	capability        string
	receipts          semantic.Receipts
}

func runtimeExecutionScopeFromContext(ctx context.Context) (RuntimeExecutionScope, bool) {
	if ctx == nil {
		return RuntimeExecutionScope{}, false
	}
	scope, ok := ctx.Value(runtimeExecutionScopeContextKey{}).(RuntimeExecutionScope)
	metadataReady := scope.view != nil ||
		scope.capability == mailboxauthorizationsemantic.CapabilityName
	return scope, ok && metadataReady && scope.receipts.CatalogRevision != 0 &&
		scope.receipts.SecurityEpoch != 0
}

// RuntimeMetadataViewFromContext is the narrow composition seam used by the
// Runtime-only MetadataService. The returned view is immutable.
func RuntimeMetadataViewFromContext(ctx context.Context) (*catalogsnapshot.RuntimeMetadataView, bool) {
	scope, ok := runtimeExecutionScopeFromContext(ctx)
	return scope.view, ok
}

// runtimeRecordAccessMetadataViewFromContext returns Related Count's
// least-privilege identity/access sibling. Admission captures it from the same
// atomically prepared serving set as the primary record-query view.
func runtimeRecordAccessMetadataViewFromContext(ctx context.Context) (*catalogsnapshot.RuntimeMetadataView, bool) {
	if ctx == nil {
		return nil, false
	}
	view, ok := ctx.Value(runtimeRecordAccessMetadataViewContextKey{}).(*catalogsnapshot.RuntimeMetadataView)
	return view, ok && view != nil
}

// runtimeRecordsMetadataViewFromContext is the composition seam for the
// Records read graph. Most calls use the admitted primary view. Related Count
// temporarily overrides it only while evaluating parent-record authority, so
// KeyPrefix identity comes from record-access/v1 while child counts continue
// to use record-query/v1.
func runtimeRecordsMetadataViewFromContext(ctx context.Context) (*catalogsnapshot.RuntimeMetadataView, bool) {
	if ctx != nil {
		if view, ok := ctx.Value(runtimeRecordsMetadataViewOverrideContextKey{}).(*catalogsnapshot.RuntimeMetadataView); ok && view != nil {
			return view, true
		}
	}
	return RuntimeMetadataViewFromContext(ctx)
}

func withRuntimeRecordsMetadataView(
	ctx context.Context,
	view *catalogsnapshot.RuntimeMetadataView,
) context.Context {
	return context.WithValue(ctx, runtimeRecordsMetadataViewOverrideContextKey{}, view)
}

// RuntimeMutationMetadataViewFromContext is private to the record-mutation
// engine graph. Flow and Approval requests receive the mutation sibling
// captured from the same atomically prepared serving set; other requests
// resolve their admitted view directly.
func RuntimeMutationMetadataViewFromContext(ctx context.Context) (*catalogsnapshot.RuntimeMetadataView, bool) {
	if ctx != nil {
		if view, ok := ctx.Value(runtimeMutationMetadataViewContextKey{}).(*catalogsnapshot.RuntimeMetadataView); ok && view != nil {
			return view, true
		}
	}
	return RuntimeMetadataViewFromContext(ctx)
}

func runtimeFlowAppMetadataFromContext(ctx context.Context) (flowAppMetadataView, bool) {
	if ctx == nil {
		return flowAppMetadataView{}, false
	}
	view, ok := ctx.Value(runtimeFlowAppMetadataContextKey{}).(flowAppMetadataView)
	return view, ok && view.OrgID != "" && view.AppInstallationID != "" &&
		view.Lineage.CatalogRevision != 0 && view.Receipts.SecurityEpoch != 0
}

func requireCatalogRevision(minimum *uint64, served uint64) error {
	if minimum == nil || *minimum <= served {
		return nil
	}
	return &semantic.Failure{
		Category: semantic.FailureStaleState, StaleAxis: semantic.StaleAxisCatalog,
		RequiredRevision: *minimum, ServedRevision: served,
	}
}

type ServingAdmission struct{ state *ServingState }

func NewServingAdmission(state *ServingState) (*ServingAdmission, error) {
	if state == nil {
		return nil, errors.New("runtime serving admission: prepared state owner is required")
	}
	return &ServingAdmission{state: state}, nil
}

func (a *ServingAdmission) Ready() bool { return a != nil && a.state != nil && a.state.Ready() }

func (a *ServingAdmission) hasFlowAppMetadataAuthority() bool {
	if a == nil || a.state == nil {
		return false
	}
	set := a.state.prepared.Load()
	if set == nil || set.byCapability[flowsemantic.CapabilityName] == nil {
		return false
	}
	return set.byCapability[catalogsemantic.AppMetadataV3CapabilityName] != nil ||
		set.byCapability[catalogsemantic.AppMetadataV2CapabilityName] != nil
}

// flowAppMetadataScope derives ExecuteFlow's internal Catalog read from the
// same atomically prepared serving set as the Flow request. The Catalog view
// capability is workload authority and is deliberately not added to the
// end-user delegation.
func (a *ServingAdmission) flowAppMetadataScope(
	trusted TrustedCall,
) (CatalogConsumerScope, string, error) {
	if a == nil || a.state == nil {
		return CatalogConsumerScope{}, "", &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	set := a.state.prepared.Load()
	if set == nil || set.byCapability[flowsemantic.CapabilityName] == nil {
		return CatalogConsumerScope{}, "", &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	for _, capability := range []string{
		catalogsemantic.AppMetadataV3CapabilityName,
		catalogsemantic.AppMetadataV2CapabilityName,
	} {
		prepared := set.byCapability[capability]
		if prepared == nil {
			continue
		}
		if trusted.orgID != prepared.selection.OrgID ||
			trusted.appInstallationID != prepared.selection.AppInstallationID ||
			trusted.callerWorkloadID != prepared.binding.WorkloadID {
			return CatalogConsumerScope{}, "", &semantic.Failure{Category: semantic.FailureForbidden}
		}
		return CatalogConsumerScope{
			OrgID:             prepared.selection.OrgID,
			AppInstallationID: prepared.selection.AppInstallationID,
			WorkloadID:        prepared.binding.WorkloadID,
			Capabilities:      []string{capability},
		}, capability, nil
	}
	return CatalogConsumerScope{}, "", &semantic.Failure{Category: semantic.FailureUnavailable}
}

func (a *ServingAdmission) 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, &semantic.Failure{Category: semantic.FailureInternal})
		}
		// Identity exchange establishes the delegation that SecurityAdmission
		// and ServingAdmission consume on the following data RPC.
		if !servingCapability(capability) {
			return handler(ctx, request)
		}
		admitted, err := a.bind(ctx, capability)
		if err != nil {
			return nil, runtimegrpc.StatusError(ctx, err)
		}
		return handler(admitted, request)
	}
}

// StreamServerInterceptor binds the exact same immutable serving view and
// receipts as unary calls. A mutation stream must not be able to collect frames
// under one Catalog/Security revision and execute them under another.
func (a *ServingAdmission) StreamServerInterceptor() grpc.StreamServerInterceptor {
	return func(server any, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		if stream == nil {
			return runtimegrpc.StatusError(context.Background(), &semantic.Failure{Category: semantic.FailureInternal})
		}
		ctx := stream.Context()
		capability, ok := runtimegrpc.Capability(ctx)
		if !ok {
			return runtimegrpc.StatusError(ctx, &semantic.Failure{Category: semantic.FailureInternal})
		}
		if !servingCapability(capability) {
			return handler(server, stream)
		}
		admitted, err := a.bind(ctx, capability)
		if err != nil {
			return runtimegrpc.StatusError(ctx, err)
		}
		return handler(server, &runtimeContextServerStream{ServerStream: stream, ctx: admitted})
	}
}

func servingCapability(capability string) bool {
	return UsesServingAdmission(capability)
}

func (a *ServingAdmission) bind(ctx context.Context, capability string) (context.Context, error) {
	if catalogConsumerCapability(capability) {
		if _, admitted := catalogConsumerScopeFromAdmission(ctx, capability); admitted {
			return ctx, nil
		}
	}
	if _, admitted := commandConsumerAuthorityFromAdmission(
		ctx,
		capability,
	); admitted {
		return ctx, nil
	}
	if a == nil || a.state == nil {
		return nil, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	set := a.state.prepared.Load()
	if set == nil {
		return nil, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	prepared := set.byCapability[capability]
	if prepared == nil {
		return nil, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	methodClass, ok := runtimegrpc.RequestMethodClass(ctx)
	if !ok {
		// Direct owner tests and internal composition probes may call bind
		// without traversing the host interceptor. They still cannot choose a
		// weaker path: only one of Runtime's private trusted context handles can
		// select the corresponding class.
		if _, trusted := trustedCallFromContext(ctx); trusted {
			methodClass = runtimegrpc.MethodClassDelegatedUserData
		} else if _, trusted := trustedWorkloadCallFromContext(ctx); trusted {
			methodClass = runtimegrpc.MethodClassWorkloadConsumer
		} else {
			return nil, &semantic.Failure{Category: semantic.FailureInternal}
		}
	}
	var callerOrgID, callerWorkloadID string
	var callerSecurityEpoch uint64
	switch methodClass {
	case runtimegrpc.MethodClassDelegatedUserData:
		trusted, trustedOK := trustedCallFromContext(ctx)
		if !trustedOK {
			return nil, &semantic.Failure{Category: semantic.FailureUnauthenticated}
		}
		if trusted.appInstallationID != prepared.selection.AppInstallationID ||
			!slices.Contains(trusted.capabilities, capability) {
			return nil, &semantic.Failure{Category: semantic.FailureForbidden}
		}
		callerOrgID, callerWorkloadID = trusted.orgID, trusted.callerWorkloadID
		callerSecurityEpoch = trusted.securityEpoch
	case runtimegrpc.MethodClassWorkloadConsumer:
		trusted, trustedOK := trustedWorkloadCallFromContext(ctx)
		if !trustedOK || trusted.capability != capability {
			return nil, &semantic.Failure{Category: semantic.FailureUnauthenticated}
		}
		callerOrgID, callerWorkloadID = trusted.orgID, trusted.callerWorkloadID
		callerSecurityEpoch = trusted.securityEpoch
	default:
		return nil, &semantic.Failure{Category: semantic.FailureInternal}
	}
	if capability != prepared.selection.RuntimeCapabilityID ||
		callerOrgID != prepared.selection.OrgID ||
		(callerWorkloadID != prepared.binding.WorkloadID && !(methodClass == runtimegrpc.MethodClassDelegatedUserData && a.state.isUserAPICaller(callerWorkloadID, capability))) {
		return nil, &semantic.Failure{Category: semantic.FailureForbidden}
	}
	if callerSecurityEpoch != prepared.selection.SecurityEpoch {
		return nil, &semantic.Failure{
			Category: semantic.FailureStaleState, StaleAxis: semantic.StaleAxisSecurity,
			RequiredRevision: prepared.selection.SecurityEpoch, ServedRevision: callerSecurityEpoch,
		}
	}
	scope := RuntimeExecutionScope{
		view: prepared.view, orgID: prepared.selection.OrgID,
		appInstallationID: prepared.selection.AppInstallationID,
		workloadID:        callerWorkloadID, capability: capability,
		receipts: prepared.receipts,
	}
	// RuntimeExecutionScope remains the API's private serving authority. Bind
	// that exact same immutable view through the neutral metadata seam as well
	// so engines reached below the Runtime API composition (records and Flow)
	// consume the admitted revision instead of falling back to process-global
	// formula vocabulary.
	ctx = metadata.WithRuntimeMetadataView(ctx, scope.view)
	ctx = context.WithValue(ctx, runtimeExecutionScopeContextKey{}, scope)
	spec, _ := servingCapabilitySpecFor(capability)
	if spec.bindsMutationEnvelope {
		var bindErr error
		ctx, bindErr = bindRecordMutationEnvelope(ctx, set)
		if bindErr != nil {
			return nil, &semantic.Failure{Category: semantic.FailureUnavailable}
		}
		if capability == flowsemantic.CapabilityName {
			if set.flowAppMetadata.OrgID == "" {
				return nil, &semantic.Failure{Category: semantic.FailureUnavailable}
			}
			ctx = context.WithValue(ctx, runtimeFlowAppMetadataContextKey{}, set.flowAppMetadata)
		}
	}
	// A record workspace may request a bounded authority projection. Bind the
	// least-privilege effective-permission sibling from the same atomic serving
	// set; the request still executes under record-view receipts and cannot
	// select another installation, revision, or principal.
	if capability == recordviewsemantic.CapabilityName {
		permissionPrepared := set.byCapability[permissionsemantic.CapabilityName]
		if permissionPrepared != nil &&
			permissionPrepared.selection.OrgID == prepared.selection.OrgID &&
			permissionPrepared.selection.AppInstallationID == prepared.selection.AppInstallationID &&
			permissionPrepared.receipts.CatalogRevision == prepared.receipts.CatalogRevision &&
			permissionPrepared.receipts.SecurityEpoch == prepared.receipts.SecurityEpoch {
			ctx = context.WithValue(ctx, runtimeEffectivePermissionMetadataViewContextKey{}, permissionPrepared.view)
		}
	}
	if capability == relatedsemantic.CapabilityName {
		accessPrepared := set.byCapability[accesssemantic.CapabilityName]
		if accessPrepared == nil ||
			accessPrepared.selection.OrgID != prepared.selection.OrgID ||
			accessPrepared.selection.AppInstallationID != prepared.selection.AppInstallationID ||
			accessPrepared.receipts.CatalogRevision != prepared.receipts.CatalogRevision ||
			accessPrepared.receipts.SecurityEpoch != prepared.receipts.SecurityEpoch {
			return nil, &semantic.Failure{Category: semantic.FailureUnavailable}
		}
		ctx = context.WithValue(ctx, runtimeRecordAccessMetadataViewContextKey{}, accessPrepared.view)
	}
	return ctx, nil
}

// bindRecordMutationEnvelope gives every Records DML entry point the same
// immutable Mutation execution authority and lifecycle services. Request
// authorization remains capability-specific, while durable after-save and
// rollup work retain the one ordinary Mutation ownership lane.
func bindRecordMutationEnvelope(ctx context.Context, set *preparedServingSet) (context.Context, error) {
	if set == nil {
		return nil, errors.New("runtime serving state: record-mutation serving set is unavailable")
	}
	mutationPrepared := set.byCapability[mutationsemantic.CapabilityName]
	if mutationPrepared == nil || mutationPrepared.view == nil {
		return nil, errors.New("runtime serving state: record-mutation sibling is unavailable")
	}
	ctx = context.WithValue(ctx, runtimeMutationMetadataViewContextKey{}, mutationPrepared.view)
	if mutationPrepared.assignmentPolicyEligibility != nil {
		ctx = managedruntime.WithAssignmentPolicyEligibilityAmbientStager(
			ctx,
			mutationPrepared.assignmentPolicyEligibility,
		)
	}
	// A durable record event carries the Mutation binding as primary and this
	// optional Flow binding as secondary. Both came from the same atomic set.
	if flowPrepared := set.byCapability[flowsemantic.CapabilityName]; flowPrepared != nil {
		ctx = metadata.WithRuntimeFlowWorkView(ctx, flowPrepared.view)
	}
	return ctx, nil
}

func runtimeViewDeclaresAssignmentPolicyEligibility(view *catalogsnapshot.RuntimeMetadataView) bool {
	if view == nil {
		return false
	}
	for _, object := range view.Objects() {
		if object != nil && object.AccessPolicy != nil && object.AccessPolicy.Lifecycle != nil &&
			object.AccessPolicy.Lifecycle.AssignmentPolicyEligibility != nil {
			return true
		}
	}
	return false
}