package runtimeapi

import (
	"context"
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"io"

	"github.com/nexuscrm/shared/pkg/models"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
)

// RecordMutationExecutor is the complete coarse command boundary. It is not a
// remote mirror of Records: the implementation owns durable command
// acceptance, CRUD/RLS/FLS, transaction execution, and reconciliation.
type RecordMutationExecutor interface {
	Mutate(context.Context, string, semantic.Request, semantic.Receipts, *models.UserSession) (semantic.Response, error)
	Status(context.Context, string, semantic.StatusRequest) (semantic.CommandShape, semantic.StatusResponse, error)
}

// RecordMutationsServer owns only neutral wire adaptation. Trusted identity,
// the exact immutable RuntimeMetadataView, and receipts are installed before
// the handler by the Runtime security and serving-admission owners.
type RecordMutationsServer struct {
	runtimev1.UnimplementedRecordMutationsServiceServer
	executor RecordMutationExecutor
}

func NewRecordMutationsServer(executor RecordMutationExecutor) (*RecordMutationsServer, error) {
	if executor == nil {
		return nil, &semantic.Failure{Category: semantic.FailureInternal}
	}
	return &RecordMutationsServer{executor: executor}, nil
}

func (s *RecordMutationsServer) MutateRecords(stream runtimev1.RecordMutationsService_MutateRecordsServer) error {
	if stream == nil {
		return mapping.ToRecordMutationStatusError(&semantic.Failure{Category: semantic.FailureInternal})
	}
	ctx := stream.Context()
	trusted, scope, receipts, err := mutationExecutionAuthority(ctx)
	if err != nil {
		return mapping.ToRecordMutationStatusError(err)
	}
	decoder := new(mapping.MutationFrameDecoder)
	for {
		frame, recvErr := stream.Recv()
		if recvErr == io.EOF {
			break
		}
		if recvErr != nil {
			return mapping.ToRecordMutationStatusError(recvErr)
		}
		if err := decoder.Add(frame); err != nil {
			return mapping.ToRecordMutationStatusError(err)
		}
	}
	request, err := decoder.Finish(ctx)
	if err != nil {
		return mapping.ToRecordMutationStatusError(err)
	}
	if request.MinCatalogRevision != nil && *request.MinCatalogRevision > receipts.CatalogRevision {
		return mapping.ToRecordMutationStatusError(&semantic.Failure{
			Category: semantic.FailureStaleState, Receipts: &receipts,
			StaleAxis: semantic.StaleAxisCatalog, RequiredRevision: *request.MinCatalogRevision,
			ServedRevision: receipts.CatalogRevision,
		})
	}
	user := trusted.UserSession()
	response, err := s.executor.Mutate(ctx, scope, request, receipts, &user)
	if err != nil {
		return mapping.ToRecordMutationStatusError(err)
	}
	wireResponse, err := mapping.ToWireMutationResponse(request, response)
	if err != nil {
		return mapping.ToRecordMutationStatusError(err)
	}
	return stream.SendAndClose(wireResponse)
}

func (s *RecordMutationsServer) GetMutationStatus(ctx context.Context, wireRequest *runtimev1.GetMutationStatusRequest) (*runtimev1.GetMutationStatusResponse, error) {
	_, scope, _, err := mutationExecutionAuthority(ctx)
	if err != nil {
		return nil, mapping.ToRecordMutationReconciliationStatusError(err)
	}
	request, err := mapping.FromWireMutationStatusRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToRecordMutationReconciliationStatusError(err)
	}
	shape, response, err := s.executor.Status(ctx, scope, request)
	if err != nil {
		return nil, mapping.ToRecordMutationReconciliationStatusError(err)
	}
	wireResponse, err := mapping.ToWireMutationStatusResponse(shape, request, response)
	if err != nil {
		return nil, mapping.ToRecordMutationReconciliationStatusError(err)
	}
	return wireResponse, nil
}

func mutationExecutionAuthority(ctx context.Context) (TrustedCall, string, semantic.Receipts, error) {
	trusted, ok := trustedCallFromContext(ctx)
	if !ok {
		return TrustedCall{}, "", semantic.Receipts{}, &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	execution, ok := runtimeExecutionScopeFromContext(ctx)
	if !ok || execution.receipts.CapabilityVersion != semantic.CapabilityName {
		return TrustedCall{}, "", semantic.Receipts{}, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	return trusted, mutationCommandScopeDigest(trusted), execution.receipts, nil
}

// mutationCommandScopeDigest is a database key derived from verified identity,
// never from caller business fields. Length-prefixing keeps the tuple injective
// if future identifier grammars admit separator bytes.
func mutationCommandScopeDigest(call TrustedCall) string {
	hash := sha256.New()
	_, _ = hash.Write([]byte("nexuscrm-runtime-record-command-scope-v1"))
	for _, value := range []string{call.orgID, call.appInstallationID, call.principalID} {
		var size [8]byte
		binary.BigEndian.PutUint64(size[:], uint64(len(value)))
		_, _ = hash.Write(size[:])
		_, _ = hash.Write([]byte(value))
	}
	return hex.EncodeToString(hash.Sum(nil))
}