package compatibility_test

import (
	"context"
	"errors"
	"net"
	"strconv"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	analyticsauthoringsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/analyticsauthoringv1"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	approvalnotificationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalnotificationflowv1"
	approvalsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalv1"
	bulkingestsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1"
	layoutauthoringsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/layoutauthoringv1"
	listviewsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/listviewv1"
	metadataadminsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/metadataadminreadv1"
	pagelayoutresolutionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/pagelayoutresolutionv1"
	permissionadminsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionadminv1"
	querysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
	accesssemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordaccessv1"
	attachmentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordattachmentsv1"
	defaultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recorddefaultsv1"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	viewsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordviewv1"
	recyclesemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recyclebinv1"
	relatedsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/relatedcountsv1"
	resourceaccesssemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/resourceaccessv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/envelope"
	commonv1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/common/v1"
	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/codes"
	"google.golang.org/grpc/credentials/insecure"
	_ "google.golang.org/grpc/encoding/gzip"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/stats"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/test/bufconn"
)

func TestGeneratedClientServerProtocolAndTransportEnvelope(t *testing.T) {
	client, service, cleanup := startTestServer(t)
	defer cleanup()

	ctx, cancel := context.WithTimeout(withProtocol(context.Background()), time.Second)
	defer cancel()
	var responseHeaders metadata.MD
	response, err := client.ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: "Issue", RecordId: "I-1", SelectedRootFields: []string{"Subject"},
	}, grpc.Header(&responseHeaders))
	if err != nil || response.GetRecordId() != "I-1" || service.calls.Load() != 1 {
		t.Fatalf("generated call = %#v, %v, calls=%d", response, err, service.calls.Load())
	}
	if encoding := responseHeaders.Get("grpc-encoding"); len(encoding) != 0 {
		t.Fatalf("v1 response compression = %#v, want identity/unset", encoding)
	}

	for _, tc := range []struct {
		name string
		ctx  context.Context
	}{
		{name: "missing capability", ctx: context.Background()},
		{name: "wrong major", ctx: metadata.NewOutgoingContext(context.Background(), metadata.Pairs(
			envelope.ProtocolMajorMetadataKey, "2", envelope.ProtocolMinorMetadataKey, "0", envelope.CapabilityMetadataKey, semantic.CapabilityName,
		))},
		{name: "wrong capability", ctx: metadata.NewOutgoingContext(context.Background(), metadata.Pairs(
			envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0", envelope.CapabilityMetadataKey, "runtime.records.other.v1",
		))},
		{name: "duplicate capability", ctx: metadata.NewOutgoingContext(context.Background(), metadata.Pairs(
			envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0", envelope.CapabilityMetadataKey, semantic.CapabilityName, envelope.CapabilityMetadataKey, semantic.CapabilityName,
		))},
	} {
		t.Run(tc.name, func(t *testing.T) {
			callCtx, callCancel := context.WithTimeout(tc.ctx, time.Second)
			defer callCancel()
			before := service.calls.Load()
			result, err := client.ReadSelectedRoot(callCtx, &runtimev1.ReadSelectedRootRequest{ObjectId: "Issue", RecordId: "I-1", SelectedRootFields: []string{"Subject"}})
			if result != nil || status.Code(err) != codes.FailedPrecondition {
				t.Fatalf("result = %#v, error = %v", result, err)
			}
			var failure *semantic.Failure
			if decoded := mapping.FromStatusError(err); !errors.As(decoded, &failure) || failure.Category != semantic.FailureIncompatible {
				t.Fatalf("decoded mismatch = %#v", decoded)
			}
			if service.calls.Load() != before {
				t.Fatal("capability mismatch reached service handler")
			}
		})
	}
}

func TestGeneratedRecordAccessClientUsesOneCoarseRPCForMixedBatch(t *testing.T) {
	client, service, cleanup := startTestServer(t)
	defer cleanup()

	ctx, cancel := context.WithTimeout(withCapability(context.Background(), accesssemantic.CapabilityName), time.Second)
	defer cancel()
	var responseHeaders metadata.MD
	response, err := client.EvaluateRecordAccess(ctx, &runtimev1.EvaluateRecordAccessRequest{
		RecordIds: []string{"W-2", "A-1", "W-2", "DENIED", "MISSING"},
	}, grpc.Header(&responseHeaders))
	if err != nil || service.accessCalls.Load() != 1 {
		t.Fatalf("generated record-access call = %#v, %v, calls=%d", response, err, service.accessCalls.Load())
	}
	if got := response.GetResults(); len(got) != 2 || got[0].GetRecordId() != "A-1" || got[1].GetRecordId() != "W-2" {
		t.Fatalf("canonical mixed results = %#v", got)
	}
	if encoding := responseHeaders.Get("grpc-encoding"); len(encoding) != 0 {
		t.Fatalf("v1 response compression = %#v, want identity/unset", encoding)
	}
	if service.calls.Load() != 0 {
		t.Fatal("record-access request reached selected-root method")
	}

	wrongCtx, wrongCancel := context.WithTimeout(withProtocol(context.Background()), time.Second)
	defer wrongCancel()
	before := service.accessCalls.Load()
	result, err := client.EvaluateRecordAccess(wrongCtx, &runtimev1.EvaluateRecordAccessRequest{RecordIds: []string{"W-2"}})
	if result != nil || status.Code(err) != codes.FailedPrecondition || service.accessCalls.Load() != before {
		t.Fatalf("wrong-capability result = %#v, error = %v, calls=%d", result, err, service.accessCalls.Load())
	}
	var failure *accesssemantic.Failure
	if decoded := mapping.FromRecordAccessStatusError(err); !errors.As(decoded, &failure) || failure.Category != accesssemantic.FailureIncompatible {
		t.Fatalf("wrong-capability decoded = %#v", decoded)
	}
}

func TestGeneratedRecordAccessEmptyBatchStillUsesOneNoopRPC(t *testing.T) {
	client, service, cleanup := startTestServer(t)
	defer cleanup()

	ctx, cancel := context.WithTimeout(withCapability(context.Background(), accesssemantic.CapabilityName), time.Second)
	defer cancel()
	response, err := client.EvaluateRecordAccess(ctx, &runtimev1.EvaluateRecordAccessRequest{})
	if err != nil || response == nil || len(response.GetResults()) != 0 || service.accessCalls.Load() != 1 {
		t.Fatalf("empty record-access call = %#v, %v, calls=%d", response, err, service.accessCalls.Load())
	}
}

func TestRecordAccessCancellationAndOperationalFailureReturnNoPartialResponse(t *testing.T) {
	t.Run("in-flight cancellation", func(t *testing.T) {
		client, service, cleanup := startTestServer(t)
		defer cleanup()
		ctx, cancel := context.WithTimeout(withCapability(context.Background(), accesssemantic.CapabilityName), time.Second)
		resultCh := make(chan *runtimev1.EvaluateRecordAccessResponse, 1)
		errCh := make(chan error, 1)
		go func() {
			result, err := client.EvaluateRecordAccess(ctx, &runtimev1.EvaluateRecordAccessRequest{RecordIds: []string{"cancel-during"}})
			resultCh <- result
			errCh <- err
		}()
		<-service.blockStarted
		cancel()
		if result := <-resultCh; result != nil {
			t.Fatalf("canceled result = %#v, want nil", result)
		}
		if err := <-errCh; status.Code(err) != codes.Canceled {
			t.Fatalf("canceled error = %v", err)
		}
	})

	t.Run("handler error with response", func(t *testing.T) {
		client, _, cleanup := startTestServer(t)
		defer cleanup()
		ctx, cancel := context.WithTimeout(withCapability(context.Background(), accesssemantic.CapabilityName), time.Second)
		defer cancel()
		result, err := client.EvaluateRecordAccess(ctx, &runtimev1.EvaluateRecordAccessRequest{RecordIds: []string{"partial-error"}})
		if result != nil || status.Code(err) != codes.Internal {
			t.Fatalf("operational failure result = %#v, error = %v", result, err)
		}
	})
}

func TestGrossOversizeFailsWithoutHandlerOrPartialResult(t *testing.T) {
	client, service, cleanup := startTestServer(t)
	defer cleanup()
	ctx, cancel := context.WithTimeout(withProtocol(context.Background()), time.Second)
	defer cancel()

	before := service.calls.Load()
	result, err := client.ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: strings.Repeat("X", envelope.MaxRequestWireBytes), RecordId: "I-1", SelectedRootFields: []string{"Subject"},
	})
	if result != nil || status.Code(err) != codes.ResourceExhausted {
		t.Fatalf("oversized request result = %#v, error = %v", result, err)
	}
	if service.calls.Load() != before {
		t.Fatal("oversized request reached service handler")
	}

	result, err = client.ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: "Issue", RecordId: "oversized-response", SelectedRootFields: []string{"Subject"},
	})
	if result != nil || status.Code(err) != codes.ResourceExhausted {
		t.Fatalf("oversized response result = %#v, error = %v", result, err)
	}
}

func TestCompressionIsNotAnAcceptedV1Path(t *testing.T) {
	if envelope.CompressionEnabled {
		t.Fatal("v1 envelope unexpectedly enables compression")
	}
	client, service, cleanup := startTestServer(t)
	defer cleanup()
	ctx, cancel := context.WithTimeout(withProtocol(context.Background()), time.Second)
	defer cancel()
	before := service.calls.Load()
	result, err := client.ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: "Issue", RecordId: "I-1", SelectedRootFields: []string{"Subject"},
	}, grpc.UseCompressor("gzip"))
	if result != nil || status.Code(err) == codes.OK {
		t.Fatalf("compressed v1 call result = %#v, error = %v", result, err)
	}
	if service.calls.Load() != before {
		t.Fatal("compressed request reached service handler")
	}
}

type testRecordsService struct {
	runtimev1.UnimplementedRecordsServiceServer
	calls        atomic.Int64
	accessCalls  atomic.Int64
	blockStarted chan struct{}
}

func (s *testRecordsService) EvaluateRecordAccess(ctx context.Context, request *runtimev1.EvaluateRecordAccessRequest) (*runtimev1.EvaluateRecordAccessResponse, error) {
	s.accessCalls.Add(1)
	response := &runtimev1.EvaluateRecordAccessResponse{
		Results: []*runtimev1.RecordAccessResult{
			{RecordId: "A-1", Read: true},
			{RecordId: "W-2", Read: true, Update: true, Delete: true},
		},
		Receipts: &commonv1.ConsistencyReceipts{
			ProtocolVersion:   &commonv1.ProtocolVersion{Major: 1},
			CapabilityVersion: &commonv1.CapabilityVersion{Name: accesssemantic.CapabilityName, Major: 1},
		},
	}
	if len(request.GetRecordIds()) == 0 {
		response.Results = nil
		return response, nil
	}
	if len(request.GetRecordIds()) == 1 {
		switch request.GetRecordIds()[0] {
		case "cancel-during":
			close(s.blockStarted)
			<-ctx.Done()
			return response, ctx.Err()
		case "partial-error":
			return response, mapping.ToRecordAccessStatusError(&accesssemantic.Failure{Category: accesssemantic.FailureInternal})
		}
	}
	return response, nil
}

func (s *testRecordsService) ReadSelectedRoot(_ context.Context, request *runtimev1.ReadSelectedRootRequest) (*runtimev1.ReadSelectedRootResponse, error) {
	s.calls.Add(1)
	value := "value"
	if request.GetRecordId() == "oversized-response" {
		value = strings.Repeat("x", envelope.MaxResponseWireBytes)
	}
	return &runtimev1.ReadSelectedRootResponse{
		RecordId: request.GetRecordId(),
		Fields: []*runtimev1.SelectedRootField{{
			Name: "Subject", DeclaredKind: runtimev1.ValueKind_VALUE_KIND_STRING,
			Value: &runtimev1.SelectedRootField_StringValue{StringValue: value},
		}},
		Receipts: &commonv1.ConsistencyReceipts{
			ProtocolVersion:   &commonv1.ProtocolVersion{Major: 1},
			CapabilityVersion: &commonv1.CapabilityVersion{Name: semantic.CapabilityName, Major: 1},
		},
	}, nil
}

func startTestServer(t *testing.T) (runtimev1.RecordsServiceClient, *testRecordsService, func()) {
	t.Helper()
	service := &testRecordsService{blockStarted: make(chan struct{})}
	connection, cleanup := startBufconnServer(t, func(server *grpc.Server) {
		runtimev1.RegisterRecordsServiceServer(server, service)
	})
	return runtimev1.NewRecordsServiceClient(connection), service, cleanup
}

// startBufconnServer runs an in-process gRPC server over bufconn with the
// standard test server and call options, registers services via register, and
// returns a dialed connection plus a cleanup function.
func startBufconnServer(t *testing.T, register func(server *grpc.Server)) (*grpc.ClientConn, func()) {
	t.Helper()
	listener := bufconn.Listen(4 * envelope.MaxResponseWireBytes)
	server := grpc.NewServer(testServerOptions()...)
	register(server)
	go func() { _ = server.Serve(listener) }()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	connection, err := grpc.DialContext(ctx, "passthrough:///bufnet",
		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultCallOptions(testDefaultCallOptions()...))
	cancel()
	if err != nil {
		server.Stop()
		_ = listener.Close()
		t.Fatal(err)
	}
	return connection, func() {
		_ = connection.Close()
		server.Stop()
		_ = listener.Close()
	}
}

// The helpers below are test-only executable handoff proof. Production gRPC
// host configuration and interception belong to nexuscrm-api-host/P3.
func withProtocol(ctx context.Context) context.Context {
	return withCapability(ctx, semantic.CapabilityName)
}

func withCapability(ctx context.Context, capability string) context.Context {
	return metadata.AppendToOutgoingContext(
		ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(semantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(semantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, capability,
	)
}

func testProtocolInterceptor(ctx context.Context, request any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
	expectedCapability := ""
	toStatus := mapping.ToStatusError
	switch info.FullMethod {
	case runtimev1.BulkIngestService_CreateBulkIngestJob_FullMethodName,
		runtimev1.BulkIngestService_ReadBulkIngestJob_FullMethodName,
		runtimev1.BulkIngestService_CloseBulkIngestJob_FullMethodName:
		expectedCapability = bulkingestsemantic.CapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
		}
	case runtimev1.ApplicationResultService_GetApplicationResult_FullMethodName:
		expectedCapability = applicationresultsemantic.CapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, applicationresultsemantic.CapabilityName)
		}
	case runtimev1.ApprovalService_ExecuteApproval_FullMethodName:
		expectedCapability = approvalsemantic.ExecuteCapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, approvalsemantic.ExecuteCapabilityName)
		}
	case runtimev1.ApprovalService_ReadApproval_FullMethodName:
		expectedCapability = approvalsemantic.ReadCapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, approvalsemantic.ReadCapabilityName)
		}
	case runtimev1.ApprovalNotificationFlowReadService_ReadApprovalNotificationFlows_FullMethodName:
		expectedCapability = approvalnotificationsemantic.CapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, approvalnotificationsemantic.CapabilityName)
		}
	case runtimev1.PageLayoutResolutionService_ResolvePageLayout_FullMethodName:
		expectedCapability = pagelayoutresolutionsemantic.CapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, pagelayoutresolutionsemantic.CapabilityName)
		}
	case runtimev1.PrivateListViewService_ReadPrivateListViews_FullMethodName,
		runtimev1.PrivateListViewService_MutatePrivateListView_FullMethodName:
		expectedCapability = listviewsemantic.CapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, listviewsemantic.CapabilityName)
		}
	case runtimev1.MetadataAdminReadService_ReadObjectAdminIndex_FullMethodName,
		runtimev1.MetadataAdminReadService_ReadLayoutAdminWorkspace_FullMethodName,
		runtimev1.MetadataAdminReadService_ReadStoredLayout_FullMethodName,
		runtimev1.MetadataAdminReadService_ReadValidationRuleAdminIndex_FullMethodName:
		expectedCapability = metadataadminsemantic.CapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, metadataadminsemantic.CapabilityName)
		}
	case runtimev1.PermissionAdministrationService_ReadPermissionAdministration_FullMethodName,
		runtimev1.PermissionAdministrationService_MutatePermissionAdministration_FullMethodName:
		expectedCapability = permissionadminsemantic.CapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, permissionadminsemantic.CapabilityName)
		}
	case runtimev1.AnalyticsAuthoringService_MutateAnalyticsDefinition_FullMethodName:
		expectedCapability = analyticsauthoringsemantic.CapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, analyticsauthoringsemantic.CapabilityName)
		}
	case runtimev1.ResourceAccessService_AccessResource_FullMethodName:
		expectedCapability = resourceaccesssemantic.CapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, resourceaccesssemantic.CapabilityName)
		}
	case runtimev1.LayoutAuthoringService_Submit_FullMethodName,
		runtimev1.LayoutAuthoringService_GetCommand_FullMethodName,
		runtimev1.LayoutAuthoringService_ReadWorkspace_FullMethodName,
		runtimev1.LayoutAuthoringService_ReadStoredLayout_FullMethodName:
		expectedCapability = layoutauthoringsemantic.CapabilityName
		toStatus = func(err error) error {
			return mapping.ToPlatformStatusError(err, layoutauthoringsemantic.CapabilityName)
		}
	case runtimev1.RecordsService_ReadSelectedRoot_FullMethodName:
		expectedCapability = semantic.CapabilityName
	case runtimev1.RecordsService_EvaluateRecordAccess_FullMethodName:
		expectedCapability = accesssemantic.CapabilityName
		toStatus = mapping.ToRecordAccessStatusError
	case runtimev1.RecordQueryService_QueryRecords_FullMethodName,
		runtimev1.RecordQueryService_QueryHistory_FullMethodName:
		expectedCapability = querysemantic.CapabilityName
		toStatus = mapping.ToQueryStatusError
	case runtimev1.RelatedRecordCountService_CountRelatedRecords_FullMethodName:
		expectedCapability = relatedsemantic.CapabilityName
		toStatus = mapping.ToRelatedCountStatusError
	case runtimev1.RecordCreateDefaultPreviewService_PreviewCreateDefaults_FullMethodName:
		expectedCapability = defaultsemantic.CapabilityName
		toStatus = mapping.ToCreateDefaultPreviewStatusError
	case runtimev1.RecycleBinReadService_ListRecycleBin_FullMethodName:
		expectedCapability = recyclesemantic.CapabilityName
		toStatus = mapping.ToRecycleBinStatusError
	case runtimev1.RecordViewReadService_ReadRecordView_FullMethodName,
		runtimev1.RecordViewReadService_ResolveRecordAttachmentSources_FullMethodName:
		expectedCapability = viewsemantic.CapabilityName
		toStatus = mapping.ToRecordViewStatusError
	case runtimev1.RecordAttachmentSourceResolutionService_ResolveAttachmentSources_FullMethodName:
		expectedCapability = attachmentsemantic.CapabilityName
		toStatus = mapping.ToRecordAttachmentSourceStatusError
	}
	values, ok := metadata.FromIncomingContext(ctx)
	compatible := expectedCapability != "" && ok &&
		exactly(values.Get(envelope.ProtocolMajorMetadataKey), "1") &&
		exactly(values.Get(envelope.ProtocolMinorMetadataKey), "0") &&
		exactly(values.Get(envelope.CapabilityMetadataKey), expectedCapability) &&
		!compressionUsed(ctx)
	if !compatible {
		return nil, toStatus(&semantic.Failure{Category: semantic.FailureIncompatible})
	}
	return handler(ctx, request)
}

func testServerOptions() []grpc.ServerOption {
	return []grpc.ServerOption{
		grpc.MaxRecvMsgSize(envelope.MaxRequestWireBytes),
		grpc.MaxSendMsgSize(envelope.MaxResponseWireBytes),
		grpc.StatsHandler(compressionStatsHandler{}),
		grpc.ChainUnaryInterceptor(testProtocolInterceptor),
	}
}

func testDefaultCallOptions() []grpc.CallOption {
	return []grpc.CallOption{
		grpc.MaxCallSendMsgSize(envelope.MaxRequestWireBytes),
		grpc.MaxCallRecvMsgSize(envelope.MaxResponseWireBytes),
		grpc.WaitForReady(false),
	}
}

func exactly(values []string, expected string) bool {
	return len(values) == 1 && values[0] == expected
}

type compressionContextKey struct{}

type compressionState struct {
	used atomic.Bool
}

type compressionStatsHandler struct{}

func (compressionStatsHandler) TagRPC(ctx context.Context, _ *stats.RPCTagInfo) context.Context {
	return context.WithValue(ctx, compressionContextKey{}, &compressionState{})
}

func (compressionStatsHandler) HandleRPC(ctx context.Context, event stats.RPCStats) {
	header, ok := event.(*stats.InHeader)
	if !ok || header.Compression == "" || header.Compression == "identity" {
		return
	}
	if state, ok := ctx.Value(compressionContextKey{}).(*compressionState); ok {
		state.used.Store(true)
	}
}

func (compressionStatsHandler) TagConn(ctx context.Context, _ *stats.ConnTagInfo) context.Context {
	return ctx
}

func (compressionStatsHandler) HandleConn(context.Context, stats.ConnStats) {}

func compressionUsed(ctx context.Context) bool {
	state, ok := ctx.Value(compressionContextKey{}).(*compressionState)
	return ok && state.used.Load()
}