package sfcrmappservice

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

	querycodec "github.com/pingcap-inc/nexuscrm-app-packs/sfcrm/querycodec"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
)

func TestGroupedQueryPreservesOverflowProbeThroughRuntime(t *testing.T) {
	ctx, cancel := context.WithTimeout(withQueryBearer(context.Background(), "test-token"), 5*time.Second)
	defer cancel()
	// The independently pinned module must still fail closed until it adopts
	// the root release; the exact-source workspace proves the upgraded path.
	probeSupported := semantic.ValidateRequest(ctx, semantic.Request{
		ObjectID: "Account", GroupBy: "Name", AggregateFetchLimit: 2001,
		Selections: []semantic.Selection{{Kind: semantic.SelectionField, Field: "Name"}},
	}) == nil
	for _, tc := range []struct {
		name   string
		rows   int
		limit  string
		probe  uint32
		status int
	}{
		{"small result", 1, "", 2001, 200},
		{"full result", 2000, "", 2001, 200},
		{"overflow detected", 2001, "", 2001, 400},
		{"explicit limit preserved", 2000, " LIMIT 2000", 2000, 200},
	} {
		t.Run(tc.name, func(t *testing.T) {
			response := semantic.Response{Done: true, Aggregate: true, TotalSize: uint64(tc.rows),
				Receipts: semantic.Receipts{ProtocolVersion: "1.0", CapabilityVersion: semantic.CapabilityName},
			}
			for i := 0; i < tc.rows; i++ {
				name := fmt.Sprintf("Group%04d", i)
				count := int64(1)
				response.Records = append(response.Records, semantic.Record{Fields: []semantic.Field{
					{Name: "Name", Value: semantic.Value{Kind: semantic.ValueString, String: &name}},
					{Name: "expr0", Value: semantic.Value{Kind: semantic.ValueInteger, Integer: &count}},
				}})
			}
			client := queryResponseTransport(t, response)
			calls := 0
			handler, err := NewRuntimeQueryHandler(func(ctx context.Context, _ string, request semantic.Request) (semantic.Response, error) {
				calls++
				if request.AggregateFetchLimit != tc.probe {
					t.Fatalf("existing probe changed: got=%d want=%d", request.AggregateFetchLimit, tc.probe)
				}
				return client.QueryRecords(ctx, request)
			}, func(context.Context, string, semantic.HistoryRequest) (semantic.HistoryResponse, error) {
				t.Fatal("grouped query reached history")
				return semantic.HistoryResponse{}, nil
			})
			if err != nil {
				t.Fatal(err)
			}
			route, _ := querycodec.Match("/services/data/v63.0/query")
			result := handler.Handle(ctx, querycodec.Request{Route: route, Query: url.Values{"q": {"SELECT Name, COUNT(Id) FROM Account GROUP BY Name" + tc.limit}}})
			wantStatus, wantCalls := tc.status, 1
			if !probeSupported && tc.probe == 2001 {
				wantStatus, wantCalls = 429, 0
			}
			body, err := json.Marshal(result.Body)
			if err != nil || result.Status != wantStatus || calls != wantCalls {
				t.Fatalf("status=%d want=%d calls=%d body=%.300s error=%v", result.Status, wantStatus, calls, body, err)
			}
			if strings.Contains(string(body), "nextRecordsUrl") {
				t.Fatal("grouped overflow must not create a continuation")
			}
			if result.Status == 400 && !strings.Contains(string(body), "EXCEEDED_ID_LIMIT") {
				t.Fatalf("sentinel lost before public overflow handling: %s", body)
			}
			if result.Status == 200 {
				var parsed struct {
					Records []any `json:"records"`
					Done    bool  `json:"done"`
				}
				if err := json.Unmarshal(body, &parsed); err != nil || !parsed.Done || len(parsed.Records) != tc.rows {
					t.Fatalf("group result changed: records=%d done=%v error=%v", len(parsed.Records), parsed.Done, err)
				}
			}
			t.Logf("probe=%d result rows=%d HTTP=%d producer calls=%d", tc.probe, tc.rows, result.Status, calls)
		})
	}
}