package runtimegrpc

import (
	"context"
	"errors"
	"io"
	"reflect"
	"sync"
	"testing"
	"time"

	identitysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/identityv1"
	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"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
)

func TestRuntimeRecordMutationClientReconcilesACommittedLostResponse(t *testing.T) {
	request := transportMutationRequest(t, "lost-response")
	want := transportMutationResponse(t, request, "record-1")
	transport := &scriptedMutationTransport{
		request: request,
		mutate:  []mutationStep{{response: want, err: &semantic.Failure{Category: semantic.FailureUnavailable}}},
		status:  []statusStep{{response: completedMutationStatus(want)}},
	}
	client := newScriptedRuntimeMutationClient(t, transport)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	got, err := client.Mutate(ctx, "end-user", request)
	if err != nil {
		t.Fatal(err)
	}
	if got.CommandDigest != want.CommandDigest || got.Results[0].RecordID != "record-1" {
		t.Fatalf("reconciled response = %#v", got)
	}
	if sequence := transport.sequenceValue(); sequence != "mutate,status" {
		t.Fatalf("transport sequence = %q", sequence)
	}
}

func TestRuntimeRecordMutationClientDoesNotReplayWhenStatusIsMissing(t *testing.T) {
	// MISSING intentionally has the same observable shape for a never-accepted,
	// expired, or digest-mismatched command. None is proof that an ambiguous
	// first attempt did not commit, so the client must preserve the ambiguity.
	request := transportMutationRequest(t, "missing-is-not-replay-proof")
	transport := &scriptedMutationTransport{
		request: request,
		mutate:  []mutationStep{{err: &semantic.Failure{Category: semantic.FailureDeadlineExceeded}}},
		status:  []statusStep{{err: &semantic.Failure{Category: semantic.FailureMissing}}},
	}
	client := newScriptedRuntimeMutationClient(t, transport)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err := client.Mutate(ctx, "end-user", request)
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureDeadlineExceeded {
		t.Fatalf("mutation error = %#v", err)
	}
	if sequence := transport.sequenceValue(); sequence != "mutate,status" {
		t.Fatalf("transport sequence = %q; MISSING authorized an unsafe replay", sequence)
	}
}

func TestRuntimeRecordMutationClientRetriesOnlyExplicitBackoffWithSameCommand(t *testing.T) {
	request := transportMutationRequest(t, "explicit-admission-backoff")
	want := transportMutationResponse(t, request, "record-after-capacity")
	admission := &semantic.Failure{
		Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitAdmission,
		Maximum: 16, RetryAfterMillis: 50,
	}
	transport := &scriptedMutationTransport{
		request: request,
		mutate:  []mutationStep{{err: admission}, {err: admission}, {response: want}},
	}
	client := newScriptedRuntimeMutationClient(t, transport)
	client.admissionRetryBudget = time.Second
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	got, err := client.Mutate(ctx, "end-user", request)
	if err != nil {
		t.Fatal(err)
	}
	if got.CommandID != request.CommandID || transport.sequenceValue() != "mutate,mutate,mutate" {
		t.Fatalf("response=%#v sequence=%q", got, transport.sequenceValue())
	}
	if len(transport.mutationRequests) != 3 {
		t.Fatalf("mutation requests = %d, want 3", len(transport.mutationRequests))
	}
	for index, replay := range transport.mutationRequests {
		if !reflect.DeepEqual(replay, request) {
			t.Fatalf("mutation request %d changed: %#v", index, replay)
		}
	}
	metrics := client.AdmissionRetryMetrics()
	if metrics.Retries != 2 || metrics.Exhausted != 0 {
		t.Fatalf("retry metrics = %+v", metrics)
	}
}

func TestRuntimeRecordMutationClientDoesNotInferRetryFromAdmissionKind(t *testing.T) {
	request := transportMutationRequest(t, "legacy-admission-is-not-instruction")
	transport := &scriptedMutationTransport{
		request: request,
		mutate: []mutationStep{{err: &semantic.Failure{
			Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitAdmission, Maximum: 16,
		}}},
	}
	client := newScriptedRuntimeMutationClient(t, transport)
	client.admissionRetryBudget = time.Second
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err := client.Mutate(ctx, "end-user", request)
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureResourceExhausted {
		t.Fatalf("mutation error = %#v", err)
	}
	if sequence := transport.sequenceValue(); sequence != "mutate" {
		t.Fatalf("transport sequence = %q; client inferred retry permission", sequence)
	}
	if metrics := client.AdmissionRetryMetrics(); metrics.Retries != 0 || metrics.Exhausted != 0 {
		t.Fatalf("retry metrics = %+v", metrics)
	}
}

func TestRuntimeRecordMutationClientKeepsAdmissionRetryDisabledWithoutOption(t *testing.T) {
	request := transportMutationRequest(t, "admission-retry-default-disabled")
	admission := &semantic.Failure{
		Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitAdmission,
		Maximum: 16, RetryAfterMillis: 50,
	}
	transport := &scriptedMutationTransport{request: request, mutate: []mutationStep{{err: admission}}}
	client := newScriptedRuntimeMutationClient(t, transport)

	_, err := client.Mutate(context.Background(), "end-user", request)
	var failure *semantic.Failure
	if !errors.As(err, &failure) || !reflect.DeepEqual(failure, admission) {
		t.Fatalf("mutation error = %#v, want admission failure", err)
	}
	if sequence := transport.sequenceValue(); sequence != "mutate" {
		t.Fatalf("transport sequence = %q; L0 enabled retry without an option", sequence)
	}
	if metrics := client.AdmissionRetryMetrics(); metrics != (RuntimeAdmissionRetryMetrics{}) {
		t.Fatalf("retry metrics = %+v", metrics)
	}
}

func TestRuntimeRecordMutationClientStopsAfterBoundedAdmissionRetries(t *testing.T) {
	request := transportMutationRequest(t, "bounded-admission-backoff")
	admission := &semantic.Failure{
		Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitAdmission,
		Maximum: 16, RetryAfterMillis: 50,
	}
	steps := make([]mutationStep, maxAdmissionRetries+1)
	for index := range steps {
		steps[index].err = admission
	}
	transport := &scriptedMutationTransport{request: request, mutate: steps}
	client := newScriptedRuntimeMutationClient(t, transport)
	client.admissionRetryBudget = 2 * time.Second
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	_, err := client.Mutate(ctx, "end-user", request)
	var failure *semantic.Failure
	if !errors.As(err, &failure) || !reflect.DeepEqual(failure, admission) {
		t.Fatalf("mutation error = %#v, want final admission failure", err)
	}
	if got := len(transport.mutationRequests); got != maxAdmissionRetries+1 {
		t.Fatalf("mutation attempts = %d, want %d", got, maxAdmissionRetries+1)
	}
	metrics := client.AdmissionRetryMetrics()
	if metrics.Retries != maxAdmissionRetries || metrics.Exhausted != 1 {
		t.Fatalf("retry metrics = %+v", metrics)
	}
}

func TestRuntimeRecordMutationClientStopsWhenAdmissionRetryBudgetIsSpent(t *testing.T) {
	request := transportMutationRequest(t, "spent-admission-budget")
	admission := &semantic.Failure{
		Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitAdmission,
		Maximum: 16, RetryAfterMillis: 50,
	}
	transport := &scriptedMutationTransport{request: request, mutate: []mutationStep{{err: admission}}}
	client := newScriptedRuntimeMutationClient(t, transport)
	client.admissionRetryBudget = time.Nanosecond

	_, err := client.Mutate(context.Background(), "end-user", request)
	var failure *semantic.Failure
	if !errors.As(err, &failure) || !reflect.DeepEqual(failure, admission) {
		t.Fatalf("mutation error = %#v, want admission failure", err)
	}
	if sequence := transport.sequenceValue(); sequence != "mutate" {
		t.Fatalf("transport sequence = %q; request retried outside its budget", sequence)
	}
	metrics := client.AdmissionRetryMetrics()
	if metrics.Retries != 0 || metrics.Exhausted != 1 {
		t.Fatalf("retry metrics = %+v", metrics)
	}
}

func TestRuntimeRecordMutationClientReconcilesAfterAdmissionRetryBecomesAmbiguous(t *testing.T) {
	request := transportMutationRequest(t, "admission-then-ambiguous")
	want := transportMutationResponse(t, request, "record-after-reconcile")
	transport := &scriptedMutationTransport{
		request: request,
		mutate: []mutationStep{
			{err: &semantic.Failure{
				Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitAdmission,
				Maximum: 16, RetryAfterMillis: 50,
			}},
			{err: &semantic.Failure{Category: semantic.FailureUnavailable}},
		},
		status: []statusStep{{response: completedMutationStatus(want)}},
	}
	client := newScriptedRuntimeMutationClient(t, transport)
	client.admissionRetryBudget = time.Second
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	got, err := client.Mutate(ctx, "end-user", request)
	if err != nil {
		t.Fatal(err)
	}
	if got.CommandID != request.CommandID || transport.sequenceValue() != "mutate,mutate,status" {
		t.Fatalf("response=%#v sequence=%q", got, transport.sequenceValue())
	}
}

func TestRuntimeRecordMutationClientReplaysAnAcceptedCommandExactlyOnce(t *testing.T) {
	request := transportMutationRequest(t, "accepted-replay")
	want := transportMutationResponse(t, request, "record-accepted")
	transport := &scriptedMutationTransport{
		request: request,
		mutate: []mutationStep{
			{err: &semantic.Failure{Category: semantic.FailureDeadlineExceeded}},
			{response: want},
		},
		status: []statusStep{{response: nonTerminalMutationStatus(t, request, semantic.CommandAccepted)}},
	}
	client := newScriptedRuntimeMutationClient(t, transport)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	got, err := client.Mutate(ctx, "end-user", request)
	if err != nil {
		t.Fatal(err)
	}
	if got.CommandDigest != want.CommandDigest || got.Results[0].RecordID != "record-accepted" {
		t.Fatalf("replayed response = %#v", got)
	}
	if sequence := transport.sequenceValue(); sequence != "mutate,status,mutate" {
		t.Fatalf("transport sequence = %q", sequence)
	}
	if transport.mutationRequests[0].CommandID != transport.mutationRequests[1].CommandID ||
		transport.mutationRequests[0].CommandID != request.CommandID {
		t.Fatalf("accepted replay changed command identity: %#v", transport.mutationRequests)
	}
}

func TestRuntimeRecordMutationClientReconcilesASecondLostResponse(t *testing.T) {
	request := transportMutationRequest(t, "second-lost-response")
	want := transportMutationResponse(t, request, "record-after-replay")
	transport := &scriptedMutationTransport{
		request: request,
		mutate: []mutationStep{
			{err: &semantic.Failure{Category: semantic.FailureUnavailable}},
			{err: &semantic.Failure{Category: semantic.FailureUnavailable}},
		},
		status: []statusStep{
			{response: nonTerminalMutationStatus(t, request, semantic.CommandAccepted)},
			{response: completedMutationStatus(want)},
		},
	}
	client := newScriptedRuntimeMutationClient(t, transport)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	got, err := client.Mutate(ctx, "end-user", request)
	if err != nil {
		t.Fatal(err)
	}
	if got.CommandDigest != want.CommandDigest || got.Results[0].RecordID != "record-after-replay" {
		t.Fatalf("second reconciliation response = %#v", got)
	}
	if sequence := transport.sequenceValue(); sequence != "mutate,status,mutate,status" {
		t.Fatalf("transport sequence = %q", sequence)
	}
}

func TestRuntimeRecordMutationClientDoesNotReplayWhenReconciliationCannotRun(t *testing.T) {
	request := transportMutationRequest(t, "reconcile-failed")
	transport := &scriptedMutationTransport{
		request: request,
		mutate:  []mutationStep{{err: &semantic.Failure{Category: semantic.FailureUnavailable}}},
		status:  []statusStep{{err: &semantic.Failure{Category: semantic.FailureUnavailable}}},
	}
	client := newScriptedRuntimeMutationClient(t, transport)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err := client.Mutate(ctx, "end-user", request)
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureUnavailable {
		t.Fatalf("mutation error = %#v", err)
	}
	if sequence := transport.sequenceValue(); sequence != "mutate,status" {
		t.Fatalf("transport sequence = %q; want no unverified replay", sequence)
	}
}

type mutationStep struct {
	response semantic.Response
	err      error
}

type statusStep struct {
	response semantic.StatusResponse
	err      error
}

type scriptedMutationTransport struct {
	mu               sync.Mutex
	request          semantic.Request
	mutate           []mutationStep
	status           []statusStep
	sequence         []string
	mutationRequests []semantic.Request
}

func (s *scriptedMutationTransport) nextMutation(request semantic.Request) (semantic.Response, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.sequence = append(s.sequence, "mutate")
	s.mutationRequests = append(s.mutationRequests, request)
	if len(s.mutate) == 0 {
		return semantic.Response{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	step := s.mutate[0]
	s.mutate = s.mutate[1:]
	return step.response, step.err
}

func (s *scriptedMutationTransport) nextStatus() (semantic.StatusResponse, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.sequence = append(s.sequence, "status")
	if len(s.status) == 0 {
		return semantic.StatusResponse{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	step := s.status[0]
	s.status = s.status[1:]
	return step.response, step.err
}

func (s *scriptedMutationTransport) sequenceValue() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	result := ""
	for i, value := range s.sequence {
		if i != 0 {
			result += ","
		}
		result += value
	}
	return result
}

func newScriptedRuntimeMutationClient(t *testing.T, transport *scriptedMutationTransport) *RuntimeRecordMutationClient {
	t.Helper()
	identity, err := NewIdentityExchangeClient(func(context.Context, *runtimev1.ExchangeRequest, ...grpc.CallOption) (*runtimev1.ExchangeResponse, error) {
		return mapping.ToWireIdentityExchangeResponse(identitysemantic.Response{
			OpaqueDelegation: "header.payload.signature", ExpiresAtUnixSecond: time.Now().Add(time.Hour).Unix(),
		})
	})
	if err != nil {
		t.Fatal(err)
	}
	mutation, err := NewMutationClient(
		func(ctx context.Context, _ ...grpc.CallOption) (grpc.ClientStreamingClient[runtimev1.MutateRecordsRequest, runtimev1.MutateRecordsResponse], error) {
			return &scriptedMutationStream{ctx: ctx, transport: transport}, nil
		},
		func(ctx context.Context, wireRequest *runtimev1.GetMutationStatusRequest, _ ...grpc.CallOption) (*runtimev1.GetMutationStatusResponse, error) {
			statusRequest, err := mapping.FromWireMutationStatusRequest(ctx, wireRequest)
			if err != nil {
				return nil, mapping.ToRecordMutationReconciliationStatusError(err)
			}
			shape, err := semantic.CommandShapeFor(transport.request)
			if err != nil {
				return nil, mapping.ToRecordMutationReconciliationStatusError(err)
			}
			response, err := transport.nextStatus()
			if err != nil {
				return nil, mapping.ToRecordMutationReconciliationStatusError(err)
			}
			return mapping.ToWireMutationStatusResponse(shape, statusRequest, response)
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	return &RuntimeRecordMutationClient{
		connection: new(grpc.ClientConn), identity: identity, mutation: mutation,
		appInstallationID: "installation-1", workloadToken: "workload", attemptTimeout: 200 * time.Millisecond,
	}
}

type scriptedMutationStream struct {
	ctx       context.Context
	transport *scriptedMutationTransport
	frames    []*runtimev1.MutateRecordsRequest
}

func (s *scriptedMutationStream) Send(frame *runtimev1.MutateRecordsRequest) error {
	s.frames = append(s.frames, frame)
	return nil
}

func (s *scriptedMutationStream) CloseAndRecv() (*runtimev1.MutateRecordsResponse, error) {
	decoder := new(mapping.MutationFrameDecoder)
	for _, frame := range s.frames {
		if err := decoder.Add(frame); err != nil {
			return nil, mapping.ToRecordMutationStatusError(err)
		}
	}
	request, err := decoder.Finish(s.ctx)
	if err != nil {
		return nil, mapping.ToRecordMutationStatusError(err)
	}
	response, stepErr := s.transport.nextMutation(request)
	if stepErr != nil {
		return nil, mapping.ToRecordMutationStatusError(stepErr)
	}
	return mapping.ToWireMutationResponse(request, response)
}

func (s *scriptedMutationStream) Header() (metadata.MD, error) { return nil, nil }
func (s *scriptedMutationStream) Trailer() metadata.MD         { return nil }
func (s *scriptedMutationStream) CloseSend() error             { return nil }
func (s *scriptedMutationStream) Context() context.Context     { return s.ctx }
func (s *scriptedMutationStream) SendMsg(message any) error {
	frame, ok := message.(*runtimev1.MutateRecordsRequest)
	if !ok {
		return errors.New("unexpected mutation frame type")
	}
	return s.Send(frame)
}
func (s *scriptedMutationStream) RecvMsg(any) error { return io.EOF }

func transportMutationRequest(t *testing.T, commandID string) semantic.Request {
	t.Helper()
	value := "value"
	return semantic.Request{
		CommandID: commandID, BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle,
		Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "WorkItem", Fields: []semantic.FieldValue{{Name: "Subject", Kind: semantic.ValueKindString, String: &value}}}},
	}
}

func transportMutationResponse(t *testing.T, request semantic.Request, recordID string) semantic.Response {
	t.Helper()
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	created := true
	return semantic.Response{
		CommandID: request.CommandID, CommandDigest: digest,
		Results:             []semantic.OperationResult{{Index: 0, Kind: semantic.OperationCreate, RecordID: recordID, Success: true, Created: &created, Errors: []semantic.OperationError{}}},
		Receipts:            semantic.Receipts{ProtocolVersion: "1.0", CatalogRevision: 7, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 10, CapabilityVersion: semantic.CapabilityName},
		OutcomeExpiresAtUTC: time.Now().Add(time.Hour).UTC().Format(time.RFC3339Nano),
	}
}

func completedMutationStatus(response semantic.Response) semantic.StatusResponse {
	receipts := response.Receipts
	return semantic.StatusResponse{
		CommandID: response.CommandID, CommandDigest: response.CommandDigest, State: semantic.CommandCompleted,
		Results: response.Results, Receipts: &receipts, OutcomeExpiresAtUTC: response.OutcomeExpiresAtUTC,
	}
}

func nonTerminalMutationStatus(t *testing.T, request semantic.Request, state semantic.CommandState) semantic.StatusResponse {
	t.Helper()
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	return semantic.StatusResponse{CommandID: request.CommandID, CommandDigest: digest, State: state}
}