package sfcrmappservice

import (
	"context"
	"encoding/json"
	"fmt"
	"net"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strings"
	"testing"
	"time"

	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	querycodec "github.com/pingcap-inc/nexuscrm-app-packs/sfcrm/querycodec"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
	"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"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/test/bufconn"
)

// This walking slice uses the production Runtime query client, actual bounded
// gRPC transport, semantic codecs, and the public HTTP query route. Only the
// database producer is replaced by a deterministic response fixture.
func TestQueryResponseLimitThroughTransportAndHTTP(t *testing.T) {
	// Prove both the independently pinned client and the jointly upgraded
	// workspace. The former must give a clear rejection rather than claim the
	// unpublished larger budget is already active.
	wideStatus := http.StatusOK
	if semantic.MaxSemanticResponseBytes == 1024*1024 {
		wideStatus = http.StatusBadRequest
	}
	for _, tc := range []struct {
		name     string
		response semantic.Response
		status   int
	}{
		{"wide page above former limit", wideQueryResponse(), wideStatus},
		{"exact advertised limit", sizedQueryResponse(t, semantic.MaxSemanticResponseBytes), http.StatusOK},
		{"one byte over", sizedQueryResponse(t, semantic.MaxSemanticResponseBytes+1), http.StatusBadRequest},
	} {
		t.Run(tc.name, func(t *testing.T) {
			size, err := semantic.SemanticResponseSize(tc.response)
			if err != nil || (tc.name == "wide page above former limit" && size <= 1024*1024) {
				t.Fatalf("invalid boundary fixture: size=%d err=%v", size, err)
			}
			client := queryResponseTransport(t, tc.response)
			service, err := NewService(ServiceConfig{
				RuntimeEvaluate: (&fakeRuntime{}).Evaluate,
				RuntimeQuery: func(ctx context.Context, _ string, request semantic.Request) (semantic.Response, error) {
					return client.QueryRecords(ctx, request)
				},
				RuntimeHistoryQuery: func(ctx context.Context, _ string, request semantic.HistoryRequest) (semantic.HistoryResponse, error) {
					return client.QueryHistory(ctx, request)
				},
				RuntimeReady:   func() bool { return true },
				RuntimeMetrics: func() runtimegrpc.RuntimeClientMetrics { return runtimegrpc.RuntimeClientMetrics{} },
				RequestTimeout: 5 * time.Second, MaxRequestBytes: 64 * 1024,
			})
			if err != nil {
				t.Fatal(err)
			}
			query := "SELECT Name FROM Account"
			if len(tc.response.Records) > 1 {
				query = "SELECT Id, A, B, C, D, E, Owner.Name FROM Account"
			}
			request := httptest.NewRequest(http.MethodGet, "/services/data/v63.0/query?q="+url.QueryEscape(query), nil)
			request.Header.Set("Authorization", "Bearer test-token")
			recorder := httptest.NewRecorder()
			service.ServeHTTP(recorder, request)
			if recorder.Code != tc.status {
				t.Fatalf("status=%d want=%d body prefix=%.300s", recorder.Code, tc.status, recorder.Body.String())
			}
			if tc.status != http.StatusOK {
				want := fmt.Sprintf(`[{"errorCode":"LIMIT_EXCEEDED","message":"Query response exceeds the encoded size limit of %d bytes. Select fewer fields or reduce the batch size using Sforce-Query-Options."}]`, semantic.MaxSemanticResponseBytes)
				if strings.TrimSpace(recorder.Body.String()) != want {
					t.Fatalf("overflow lost actionable limit: %s", recorder.Body.String())
				}
				return
			}
			var result struct {
				Done      bool             `json:"done"`
				Records   []map[string]any `json:"records"`
				TotalSize int              `json:"totalSize"`
			}
			if err := json.Unmarshal(recorder.Body.Bytes(), &result); err != nil {
				t.Fatal(err)
			}
			if !result.Done || len(result.Records) != len(tc.response.Records) || result.TotalSize != len(tc.response.Records) {
				t.Fatalf("response truncated: done=%v records=%d total=%d", result.Done, len(result.Records), result.TotalSize)
			}
			if len(result.Records) == 1 {
				if result.Records[0]["Name"] != *tc.response.Records[0].Fields[1].Value.String {
					t.Fatal("boundary string was truncated")
				}
			} else {
				for _, record := range result.Records {
					owner, ok := record["Owner"].(map[string]any)
					if !ok || owner["Name"] != strings.Repeat("x", 48) || record["E"] != strings.Repeat("x", 48) {
						t.Fatal("wide page lost selected or relationship fields")
					}
				}
			}
			t.Logf("complete response: records=%d semantic bytes=%d HTTP bytes=%d", len(result.Records), size, recorder.Body.Len())
		})
	}
}

func TestQueryLimitMessagesPreservePeerMaximumAndDistinguishOverload(t *testing.T) {
	for _, history := range []bool{false, true} {
		for _, maximum := range []uint64{1024 * 1024, semantic.MaxSemanticResponseBytes} {
			err := runtimeQueryFailure(&semantic.Failure{Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitSemanticResponse, Maximum: maximum}, history)
			failure := err.(*querycodec.Failure)
			if failure.Category != querycodec.FailureResponseTooLarge || !strings.Contains(failure.Message, fmt.Sprintf("%d bytes", maximum)) {
				t.Fatalf("serving peer's limit lost: %#v", failure)
			}
			if history && (!strings.Contains(failure.Message, "SOQL LIMIT") || strings.Contains(failure.Message, "Sforce-Query-Options")) {
				t.Fatalf("history advice must use a supported control: %s", failure.Message)
			}
		}
	}
	overload := runtimeQueryFailure(&semantic.Failure{Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitAdmission, Maximum: 4}, false).(*querycodec.Failure)
	if !strings.Contains(overload.Message, "concurrent request limit") || !strings.Contains(overload.Message, "Retry later") || strings.Contains(overload.Message, "bytes") {
		t.Fatalf("overload presented as payload size failure: %s", overload.Message)
	}
	unknown := runtimeQueryFailure(&semantic.Failure{Category: semantic.FailureResourceExhausted, LimitKind: "private query text", Maximum: 999}, false).(*querycodec.Failure)
	if unknown.Message != "Request exceeds a supported limit" {
		t.Fatalf("unreviewed limit detail disclosed: %s", unknown.Message)
	}
}

func sizedQueryResponse(t *testing.T, target int) semantic.Response {
	t.Helper()
	text, id := "", "001000000000000001"
	response := semantic.Response{TotalSize: 1, Done: true,
		Receipts: semantic.Receipts{ProtocolVersion: "1.0", CapabilityVersion: semantic.CapabilityName},
		Records: []semantic.Record{{ObjectID: "Account", Fields: []semantic.Field{
			{Name: "Id", Value: semantic.Value{Kind: semantic.ValueReference, Reference: &id}},
			{Name: "Name", Value: semantic.Value{Kind: semantic.ValueString, String: &text}},
		}}},
	}
	size, err := semantic.SemanticResponseSize(response)
	if err != nil {
		t.Fatal(err)
	}
	text = strings.Repeat("x", target-size)
	return response
}

func wideQueryResponse() semantic.Response {
	text := strings.Repeat("x", 48)
	response := semantic.Response{TotalSize: 2000, Done: true,
		Receipts: semantic.Receipts{ProtocolVersion: "1.0", CapabilityVersion: semantic.CapabilityName},
	}
	for i := 0; i < 2000; i++ {
		record := semantic.Record{ObjectID: "Account"}
		for _, name := range []string{"A", "B", "C", "D", "E"} {
			record.Fields = append(record.Fields, semantic.Field{Name: name, Value: semantic.Value{Kind: semantic.ValueString, String: &text}})
		}
		id := fmt.Sprintf("001%015d", i+1)
		record.Fields = append(record.Fields, semantic.Field{Name: "Id", Value: semantic.Value{Kind: semantic.ValueReference, Reference: &id}})
		record.Fields = append(record.Fields, semantic.Field{Name: "Owner", Value: semantic.Value{Kind: semantic.ValueObject, Object: &semantic.Record{
			ObjectID: "User", Fields: []semantic.Field{{Name: "Name", Value: semantic.Value{Kind: semantic.ValueString, String: &text}}},
		}}})
		response.Records = append(response.Records, record)
	}
	return response
}

type queryResponseServer struct {
	runtimev1.UnimplementedRecordQueryServiceServer
	response semantic.Response
	failure  *semantic.Failure
	query    func(context.Context, semantic.Request) (semantic.Response, error)
}

func (s *queryResponseServer) QueryRecords(ctx context.Context, request *runtimev1.QueryRecordsRequest) (*runtimev1.QueryRecordsResponse, error) {
	if s.failure != nil {
		return nil, mapping.ToQueryStatusError(s.failure)
	}
	decoded, err := mapping.FromWireQueryRecordsRequest(ctx, request)
	if err != nil {
		return nil, mapping.ToQueryStatusError(err)
	}
	result := s.response
	if s.query != nil {
		result, err = s.query(ctx, decoded)
		if err != nil {
			return nil, mapping.ToQueryStatusError(err)
		}
	}
	response, err := mapping.ToWireQueryRecordsResponse(result)
	return response, mapping.ToQueryStatusError(err)
}

func (s *queryResponseServer) QueryHistory(context.Context, *runtimev1.QueryHistoryRequest) (*runtimev1.QueryHistoryResponse, error) {
	return nil, mapping.ToQueryStatusError(s.failure)
}

func queryResponseTransport(t *testing.T, response semantic.Response, failures ...*semantic.Failure) *runtimegrpc.QueryClient {
	t.Helper()
	producer := &queryResponseServer{response: response}
	if len(failures) > 0 {
		producer.failure = failures[0]
	}
	return queryResponseTransportProducer(t, producer)
}

func queryResponseTransportProducer(t *testing.T, producer *queryResponseServer) *runtimegrpc.QueryClient {
	t.Helper()
	listener := bufconn.Listen(4 * envelope.MaxResponseWireBytes)
	server := grpc.NewServer(grpc.MaxRecvMsgSize(envelope.MaxRequestWireBytes), grpc.MaxSendMsgSize(envelope.MaxResponseWireBytes))
	runtimev1.RegisterRecordQueryServiceServer(server, producer)
	go func() { _ = server.Serve(listener) }()
	t.Cleanup(func() { server.Stop(); _ = listener.Close() })
	connection, err := grpc.NewClient("passthrough:///query-limit", grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = connection.Close() })
	generated := runtimev1.NewRecordQueryServiceClient(connection)
	client, err := runtimegrpc.NewQueryClient(generated.QueryRecords, generated.QueryHistory)
	if err != nil {
		t.Fatal(err)
	}
	return client
}

func TestQuerySizeAndAdmissionFailuresHaveDistinctHTTPStatus(t *testing.T) {
	for _, query := range []string{"SELECT Id FROM Account", "SELECT Id FROM AccountHistory WHERE AccountId = '001parent' LIMIT 5"} {
		for _, tc := range []struct {
			name    string
			failure *semantic.Failure
			status  int
			advice  string
		}{
			{"response size", &semantic.Failure{Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitSemanticResponse, Maximum: semantic.MaxSemanticResponseBytes}, 400, "bytes"},
			{"admission", &semantic.Failure{Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitAdmission, Maximum: 4}, 429, "Retry later"},
		} {
			t.Run(query+"/"+tc.name, func(t *testing.T) {
				client := queryResponseTransport(t, semantic.Response{}, tc.failure)
				calls := 0
				handler, err := NewRuntimeQueryHandler(func(ctx context.Context, _ string, request semantic.Request) (semantic.Response, error) {
					calls++
					return client.QueryRecords(ctx, request)
				}, func(ctx context.Context, _ string, request semantic.HistoryRequest) (semantic.HistoryResponse, error) {
					calls++
					return client.QueryHistory(ctx, request)
				})
				if err != nil {
					t.Fatal(err)
				}
				route, _ := querycodec.Match("/services/data/v63.0/query")
				ctx, cancel := context.WithTimeout(withQueryBearer(context.Background(), "test-token"), 5*time.Second)
				defer cancel()
				result := handler.Handle(ctx, querycodec.Request{Route: route, Query: url.Values{"q": {query}}})
				body, err := json.Marshal(result.Body)
				if err != nil || result.Status != tc.status || calls != 1 || !strings.Contains(string(body), `"errorCode":"LIMIT_EXCEEDED"`) || !strings.Contains(string(body), tc.advice) {
					t.Fatalf("status=%d want=%d calls=%d body=%s err=%v", result.Status, tc.status, calls, body, err)
				}
			})
		}
	}
}