package runtimegrpc

import (
	"context"
	"errors"

	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/envelope"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
	"google.golang.org/grpc"
)

// MutationStreamInvoker and MutationStatusInvoker are the generated mutation
// method shapes. Function values remain directly nil-checkable and keep
// connection credentials, resolution, and lifecycle with the process owner.
type MutationStreamInvoker func(context.Context, ...grpc.CallOption) (grpc.ClientStreamingClient[runtimev1.MutateRecordsRequest, runtimev1.MutateRecordsResponse], error)
type MutationStatusInvoker func(context.Context, *runtimev1.GetMutationStatusRequest, ...grpc.CallOption) (*runtimev1.GetMutationStatusResponse, error)

// MutationClient owns only app-blind framing, protocol metadata, envelopes,
// and semantic/wire validation. Command identity and product projection remain
// with L1; Runtime remains the sole outcome authority.
type MutationClient struct {
	mutate MutationStreamInvoker
	status MutationStatusInvoker
}

func NewMutationClient(mutate MutationStreamInvoker, status MutationStatusInvoker) (*MutationClient, error) {
	if mutate == nil || status == nil {
		return nil, errors.New("runtimegrpc: mutation stream and status invokers are required")
	}
	return &MutationClient{mutate: mutate, status: status}, nil
}

func (c *MutationClient) MutateRecords(ctx context.Context, request semantic.Request) (semantic.Response, error) {
	if c == nil || c.mutate == nil {
		return semantic.Response{}, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	frames, err := mapping.EncodeMutationFrames(ctx, request)
	if err != nil {
		return semantic.Response{}, err
	}
	stream, err := c.mutate(withClientProtocol(ctx, semantic.ProtocolMajor, semantic.ProtocolMinor, semantic.CapabilityName),
		grpc.MaxCallSendMsgSize(envelope.MaxRequestWireBytes),
		grpc.MaxCallRecvMsgSize(envelope.MaxResponseWireBytes),
		grpc.WaitForReady(false),
	)
	if err != nil {
		return semantic.Response{}, mapping.FromRecordMutationStatusError(err)
	}
	for _, frame := range frames {
		if err := stream.Send(frame); err != nil {
			return semantic.Response{}, mapping.FromRecordMutationStatusError(err)
		}
	}
	wireResponse, err := stream.CloseAndRecv()
	if err != nil {
		return semantic.Response{}, mapping.FromRecordMutationStatusError(err)
	}
	return mapping.FromWireMutationResponse(request, wireResponse)
}

func (c *MutationClient) GetMutationStatus(ctx context.Context, shape semantic.CommandShape, request semantic.StatusRequest) (semantic.StatusResponse, error) {
	if c == nil || c.status == nil {
		return semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	wireRequest, err := mapping.ToWireMutationStatusRequest(ctx, request)
	if err != nil {
		return semantic.StatusResponse{}, err
	}
	wireResponse, err := c.status(withClientProtocol(ctx, semantic.ProtocolMajor, semantic.ProtocolMinor, semantic.CapabilityName), wireRequest,
		grpc.MaxCallSendMsgSize(envelope.MaxRequestWireBytes),
		grpc.MaxCallRecvMsgSize(envelope.MaxResponseWireBytes),
		grpc.WaitForReady(false),
	)
	if err != nil {
		return semantic.StatusResponse{}, mapping.FromRecordMutationReconciliationStatusError(err)
	}
	return mapping.FromWireMutationStatusResponse(shape, request, wireResponse)
}