package runtimeapi

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	flowcommandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv1"
	flowcommandsemanticv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
	flowcommandsemanticv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
	recordsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
)

type CommandAcceptanceEngine interface {
	Accept(context.Context, persistence.RuntimePlatformCommandAcceptance) (persistence.RuntimePlatformCommandRow, bool, error)
}

type CommandAcceptanceServer struct {
	runtimev1.UnimplementedCommandAcceptanceServiceServer
	engine CommandAcceptanceEngine
}

func NewCommandAcceptanceServer(engine CommandAcceptanceEngine) (*CommandAcceptanceServer, error) {
	if engine == nil {
		return nil, errors.New("runtime command acceptance: engine is required")
	}
	return &CommandAcceptanceServer{engine: engine}, nil
}

func (s *CommandAcceptanceServer) AcceptCommand(
	ctx context.Context,
	wireRequest *runtimev1.AcceptCommandRequest,
) (*runtimev1.AcceptCommandResponse, error) {
	request, err := mapping.FromWireAcceptCommandRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, commandsemantic.CapabilityName)
	}
	scope, ok := runtimeExecutionScopeFromContext(ctx)
	trusted, trustedOK := trustedWorkloadCallFromContext(ctx)
	if !ok || !trustedOK {
		return nil, mapping.ToPlatformStatusError(
			&recordsemantic.Failure{Category: recordsemantic.FailureUnavailable},
			commandsemantic.CapabilityName,
		)
	}
	if err := requireCatalogRevision(request.MinCatalogRevision, scope.receipts.CatalogRevision); err != nil {
		return nil, mapping.ToPlatformStatusError(err, commandsemantic.CapabilityName)
	}
	if trusted.orgID != scope.orgID || trusted.callerWorkloadID != scope.workloadID ||
		trusted.capability != commandsemantic.CapabilityName {
		return nil, mapping.ToPlatformStatusError(
			&recordsemantic.Failure{Category: recordsemantic.FailureForbidden},
			commandsemantic.CapabilityName,
		)
	}
	if scope.view == nil {
		return nil, mapping.ToPlatformStatusError(
			&recordsemantic.Failure{Category: recordsemantic.FailureUnavailable},
			commandsemantic.CapabilityName,
		)
	}
	if !commandRequestHasExecutionAuthority(ctx, scope, request) {
		return nil, mapping.ToPlatformStatusError(
			&recordsemantic.Failure{
				Category:         recordsemantic.FailureInvalidRequest,
				InvalidComponent: commonsemantic.InvalidComponentCommandData,
			},
			commandsemantic.CapabilityName,
		)
	}
	binding, err := catalogsnapshot.RuntimeWorkBindingFromView(scope.view)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(
			&recordsemantic.Failure{Category: recordsemantic.FailureInternal},
			commandsemantic.CapabilityName,
		)
	}
	encodedBinding, err := catalogsnapshot.EncodeRuntimeWorkBinding(binding)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(
			&recordsemantic.Failure{Category: recordsemantic.FailureInternal},
			commandsemantic.CapabilityName,
		)
	}
	executionScopeDigest, err := (persistence.RuntimeWorkScope{
		OrgID: scope.orgID, AppInstallationID: scope.appInstallationID,
		RuntimeCapabilityID: commandsemantic.CapabilityName,
	}).Digest()
	if err != nil {
		return nil, mapping.ToPlatformStatusError(
			&recordsemantic.Failure{Category: recordsemantic.FailureInternal},
			commandsemantic.CapabilityName,
		)
	}
	authority := runtimeCommandAuthority{
		orgID: scope.orgID, appInstallationID: scope.appInstallationID,
		workloadID: scope.workloadID, metadataBinding: encodedBinding,
		executionScopeDigest: executionScopeDigest,
	}
	receipts := receiptsForCapability(scope.receipts, commandsemantic.CapabilityName)
	row, duplicate, err := acceptRuntimePlatformCommand(
		ctx, s.engine, authority, commandsemantic.CapabilityName, receipts, request,
	)
	if err != nil {
		if ctxErr := ctx.Err(); ctxErr != nil {
			return nil, mapping.ToPlatformStatusError(ctxErr, commandsemantic.CapabilityName)
		}
		failure := &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
		switch {
		case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
			failure.Category = recordsemantic.FailureUnavailable
		case errors.Is(err, persistence.ErrRuntimePlatformCommandConflict):
			failure.Category = recordsemantic.FailureInvalidRequest
		case errors.Is(err, persistence.ErrRuntimePlatformCommandServingBindingChanged):
			failure = servingBindingFailure(row.Receipts, receipts)
		}
		return nil, mapping.ToPlatformStatusError(failure, commandsemantic.CapabilityName)
	}
	wireResponse, err := mapping.ToWireAcceptCommandResponse(request, commandsemantic.Response{
		CommandID: request.CommandID, Duplicate: duplicate, Receipts: receipts,
	})
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, commandsemantic.CapabilityName)
	}
	return wireResponse, nil
}

// commandRequestHasExecutionAuthority keeps ordinary kind/version commands on
// their published v1 behavior. The launch Runtime Flow command is
// definition-selecting, so its exact payload identity must also be authorized
// for the admitted workload by the same immutable command view before a
// durable command owner is created.
func commandRequestHasExecutionAuthority(
	ctx context.Context,
	scope RuntimeExecutionScope,
	request commandsemantic.Request,
) bool {
	if scope.view == nil ||
		!scope.view.AllowsCommand(
			request.CommandKind,
			request.PayloadFormatVersion,
		) {
		return false
	}
	current, consumerAdmission := commandConsumerAuthorityFromAdmission(
		ctx,
		commandsemantic.CapabilityName,
	)
	if request.CommandKind != flowcommandsemantic.CommandKind {
		// Handler-owned N-1 commands retain their published kind/version
		// behavior. The independent Integration consumer may submit only the
		// definition-selecting command whose exact Catalog binding can be
		// checked before durable ownership.
		return !consumerAdmission
	}
	definitionRef, definitionVersion, valid := runtimeFlowCommandIdentity(request)
	if !valid {
		return false
	}
	contextNamespace, authorized := scope.view.CommandExecutionContext(
		request.CommandKind,
		request.PayloadFormatVersion,
		scope.workloadID,
		definitionRef,
		definitionVersion,
	)
	if !authorized {
		return false
	}
	if !consumerAdmission {
		return true
	}
	for _, binding := range current.Bindings {
		command := binding.RuntimeCommand
		if command != nil &&
			command.CommandKind == request.CommandKind &&
			command.PayloadFormatVersion == request.PayloadFormatVersion &&
			command.DefinitionRef == definitionRef &&
			command.DefinitionVersion == definitionVersion &&
			binding.OwnerPack == contextNamespace {
			return true
		}
	}
	return false
}

// runtimeFlowCommandIdentity decodes only the published payload versions that
// the Runtime Worker artifact owns. Version selection remains Catalog-owned:
// this helper extracts the exact definition identity but cannot authorize a
// version absent from the immutable Runtime and Integration projections.
func runtimeFlowCommandIdentity(
	request commandsemantic.Request,
) (definitionRef string, definitionVersion string, valid bool) {
	switch request.PayloadFormatVersion {
	case flowcommandsemantic.PayloadFormatVersion:
		payload, err := flowcommandsemantic.Decode(request.Payload)
		if err != nil {
			return "", "", false
		}
		return payload.DefinitionRef, payload.DefinitionVersion, true
	case flowcommandsemanticv2.PayloadFormatVersion:
		payload, err := flowcommandsemanticv2.Decode(request.Payload)
		if err != nil {
			return "", "", false
		}
		return payload.DefinitionRef, payload.DefinitionVersion, true
	case flowcommandsemanticv3.PayloadFormatVersion:
		payload, err := flowcommandsemanticv3.Decode(request.Payload)
		if err != nil {
			return "", "", false
		}
		return payload.ExecutionRef.DefinitionRef,
			payload.ExecutionRef.DefinitionVersion,
			true
	default:
		return "", "", false
	}
}

type runtimeCommandAuthority struct {
	orgID                string
	appInstallationID    string
	workloadID           string
	metadataBinding      string
	executionScopeDigest string
}

func acceptRuntimePlatformCommand(
	ctx context.Context,
	engine CommandAcceptanceEngine,
	authority runtimeCommandAuthority,
	authorityCapability string,
	receipts commonsemantic.Receipts,
	request commandsemantic.Request,
) (persistence.RuntimePlatformCommandRow, bool, error) {
	receiptPayload, err := json.Marshal(receipts)
	if err != nil {
		return persistence.RuntimePlatformCommandRow{}, false, err
	}
	scopeDigest, err := (persistence.RuntimePlatformCommandTrustedScope{
		OrgID: authority.orgID, AppInstallationID: authority.appInstallationID,
		CallerWorkloadID: authority.workloadID, RuntimeCapabilityID: authorityCapability,
	}).Digest()
	if err != nil {
		return persistence.RuntimePlatformCommandRow{}, false, err
	}
	requestPayload, err := json.Marshal(struct {
		CommandID            string `json:"commandId"`
		CommandKind          string `json:"commandKind"`
		PartitionKey         string `json:"partitionKey,omitempty"`
		PayloadFormatVersion uint32 `json:"payloadFormatVersion"`
		PayloadDigest        string `json:"payloadDigest"`
	}{
		CommandID: request.CommandID, CommandKind: request.CommandKind,
		PartitionKey: request.PartitionKey, PayloadFormatVersion: request.PayloadFormatVersion,
		PayloadDigest: request.PayloadDigest,
	})
	if err != nil {
		return persistence.RuntimePlatformCommandRow{}, false, err
	}
	requestSum := sha256.Sum256(requestPayload)
	return engine.Accept(ctx, persistence.RuntimePlatformCommandAcceptance{
		RowID: deterministicFileID(scopeDigest, request.CommandID, request.IdempotencyKey),
		OrgID: authority.orgID, AppInstallationID: authority.appInstallationID,
		CallerWorkloadID: authority.workloadID, RuntimeCapabilityID: authorityCapability,
		MetadataBinding:      authority.metadataBinding,
		ExecutionScopeDigest: authority.executionScopeDigest,
		ScopeDigest:          scopeDigest, CommandID: request.CommandID,
		IdempotencyKey: request.IdempotencyKey, CommandKind: request.CommandKind,
		PartitionKey: request.PartitionKey, PayloadFormatVersion: request.PayloadFormatVersion,
		RequestDigest: "sha256:" + hex.EncodeToString(requestSum[:]),
		Payload:       append([]byte(nil), request.Payload...), Receipts: receiptPayload,
	})
}

func servingBindingFailure(storedPayload []byte, current commonsemantic.Receipts) *recordsemantic.Failure {
	var stored commonsemantic.Receipts
	if json.Unmarshal(storedPayload, &stored) != nil {
		return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
	}
	if stored.SecurityEpoch != current.SecurityEpoch {
		return &recordsemantic.Failure{
			Category: recordsemantic.FailureStaleState, StaleAxis: recordsemantic.StaleAxisSecurity,
			RequiredRevision: current.SecurityEpoch, ServedRevision: stored.SecurityEpoch,
		}
	}
	if stored.CatalogRevision != current.CatalogRevision {
		return &recordsemantic.Failure{
			Category: recordsemantic.FailureStaleState, StaleAxis: recordsemantic.StaleAxisCatalog,
			RequiredRevision: current.CatalogRevision, ServedRevision: stored.CatalogRevision,
		}
	}
	return &recordsemantic.Failure{Category: recordsemantic.FailureIncompatible}
}

var _ runtimev1.CommandAcceptanceServiceServer = (*CommandAcceptanceServer)(nil)