package runtimegrpc

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	bulk "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1"
	common "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	identity "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/identityv1"
	layout "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/layoutauthoringv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

func identitySecurityRejection() error {
	return mapping.ToPlatformStatusError(&common.Failure{
		Category: common.FailureStaleState, StaleAxis: common.StaleAxisSecurity,
		RequiredRevision: 2, ServedRevision: 1,
	}, layout.CapabilityName)
}

func TestRuntimeIdentityRejectionRefreshesOnNextCallWithoutReplayingRPC(t *testing.T) {
	target, server, transport, cleanup := startPlatformRuntimeTestServer(t)
	defer cleanup()
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	client, err := dialPlatformRuntimeWithCredentials(ctx, RuntimeClientConfig{
		Target: target, AppInstallationID: "installation-1", WorkloadToken: "workload-secret", AttemptTimeout: time.Second,
	}, transport)
	if err != nil {
		t.Fatal(err)
	}
	defer client.Close()
	request := layout.GetCommandRequest{CommandID: "layout-command-1", RequestDigest: "sha256:" + strings.Repeat("a", 64)}
	if _, err := client.GetLayoutAuthoringCommand(ctx, "user-secret", request); err != nil {
		t.Fatal(err)
	}
	server.layoutRejectSecurityOnce.Store(true)
	_, err = client.GetLayoutAuthoringCommand(ctx, "user-secret", request)
	var failure *common.Failure
	if !errors.As(err, &failure) || failure.Category != common.FailureStaleState || failure.StaleAxis != common.StaleAxisSecurity {
		t.Fatalf("original rejection changed: %v", err)
	}
	if server.identityCalls.Load() != 1 || server.layoutAuthoringStatuses.Load() != 2 {
		t.Fatal("client replayed the rejected RPC or proactively exchanged identity")
	}
	for range 2 {
		if _, err := client.GetLayoutAuthoringCommand(ctx, "user-secret", request); err != nil {
			t.Fatal(err)
		}
	}
	if server.identityCalls.Load() != 2 || server.layoutAuthoringStatuses.Load() != 4 {
		t.Fatalf("identity/status calls = %d/%d, want 2/4", server.identityCalls.Load(), server.layoutAuthoringStatuses.Load())
	}
}

func TestIdentityRejectionCoalescesRefreshAndPreservesNewerAndUnrelatedEntries(t *testing.T) {
	coordinator := &identityExchangeCoordinator{bursts: newIdentityBurstCache()}
	request := identity.Request{AppInstallationID: "installation-1", RequestedCapabilities: []string{layout.CapabilityName}}
	var calls atomic.Int64
	exchange := func(context.Context, string, identity.Request) (identity.Response, error) {
		return identity.Response{OpaqueDelegation: fmt.Sprintf("delegation-%d", calls.Add(1)), ExpiresAtUnixSecond: time.Now().Add(time.Minute).Unix()}, nil
	}
	old, _ := coordinator.exchange(t.Context(), "bearer", request, exchange)
	unrelated, _ := coordinator.exchange(t.Context(), "other-bearer", request, exchange)
	oldContext := withRuntimeDelegation(t.Context(), old.OpaqueDelegation)
	coordinator.observeRejection(oldContext, identitySecurityRejection())
	const callers = 12
	var wait sync.WaitGroup
	for range callers {
		wait.Add(1)
		go func() {
			defer wait.Done()
			response, err := coordinator.exchange(t.Context(), "bearer", request, exchange)
			if err != nil || response.OpaqueDelegation != "delegation-3" {
				t.Errorf("refresh = %#v, %v", response, err)
			}
		}()
	}
	wait.Wait()
	coordinator.observeRejection(oldContext, identitySecurityRejection())
	fresh, _ := coordinator.exchange(t.Context(), "bearer", request, exchange)
	stillUnrelated, _ := coordinator.exchange(t.Context(), "other-bearer", request, exchange)
	if calls.Load() != 3 || fresh.OpaqueDelegation != "delegation-3" || stillUnrelated.OpaqueDelegation != unrelated.OpaqueDelegation {
		t.Fatalf("late rejection evicted newer/unrelated authority; exchanges=%d", calls.Load())
	}
}

func TestIdentityRejectionDoesNotEvictForUnrelatedFailures(t *testing.T) {
	for name, err := range map[string]error{
		"success":   nil,
		"transport": status.Error(codes.Unavailable, "unavailable"),
		"untyped":   status.Error(codes.FailedPrecondition, "stale"),
		"catalog":   mapping.ToPlatformStatusError(&common.Failure{Category: common.FailureStaleState, StaleAxis: common.StaleAxisCatalog, RequiredRevision: 2, ServedRevision: 1}, layout.CapabilityName),
		"forbidden": mapping.ToPlatformStatusError(&common.Failure{Category: common.FailureForbidden}, layout.CapabilityName),
	} {
		t.Run(name, func(t *testing.T) {
			coordinator := &identityExchangeCoordinator{bursts: newIdentityBurstCache()}
			request := identity.Request{AppInstallationID: "installation", RequestedCapabilities: []string{layout.CapabilityName}}
			calls := 0
			exchange := func(context.Context, string, identity.Request) (identity.Response, error) {
				calls++
				return identity.Response{OpaqueDelegation: "delegation", ExpiresAtUnixSecond: time.Now().Add(time.Minute).Unix()}, nil
			}
			_, _ = coordinator.exchange(t.Context(), "bearer", request, exchange)
			coordinator.observeRejection(withRuntimeDelegation(t.Context(), "delegation"), err)
			_, _ = coordinator.exchange(t.Context(), "bearer", request, exchange)
			if calls != 1 {
				t.Fatalf("unrelated failure evicted identity: %d exchanges", calls)
			}
		})
	}
}

func TestRuntimeIdentityRejectionOnUploadSendReadsStatusBeforeCancel(t *testing.T) {
	target, server, transport, cleanup := startPlatformRuntimeTestServer(t)
	defer cleanup()
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	client, err := dialPlatformRuntimeWithCredentials(ctx, RuntimeClientConfig{Target: target, AppInstallationID: "installation-1", WorkloadToken: "workload-secret", AttemptTimeout: time.Second}, transport)
	if err != nil {
		t.Fatal(err)
	}
	defer client.Close()
	server.bulkRejectRelease = make(chan struct{})
	server.bulkRejectSecurityOnce.Store(true)
	upload, err := client.BeginBulkIngestUpload(ctx, "user-secret", bulk.UploadHeader{JobID: "job-1", ContentLength: 1, ContentLengthKnown: true})
	if err != nil {
		t.Fatal(err)
	}
	defer upload.Cancel()
	close(server.bulkRejectRelease)
	// Synchronize with the real server's terminal rejection. Send now sees EOF,
	// not the typed status; the upload must receive that status before cancel.
	_, _ = upload.stream.Header()
	err = upload.Send([]byte("a"))
	var failure *common.Failure
	if !errors.As(err, &failure) || failure.Category != common.FailureStaleState || failure.StaleAxis != common.StaleAxisSecurity {
		t.Fatalf("lost upload rejection: %v", err)
	}
	if server.identityCalls.Load() != 1 || server.bulkUploadCalls.Load() != 1 {
		t.Fatal("rejected upload was replayed")
	}
	if _, err := client.CreateBulkIngestJob(ctx, "user-secret", bulk.CreateRequest{ObjectID: "Account", Operation: bulk.OperationInsert}); err != nil {
		t.Fatal(err)
	}
	if server.identityCalls.Load() != 2 || server.bulkCreateCalls.Load() != 1 {
		t.Fatal("next upload lifecycle request reused rejected identity")
	}
}