package runtimeapi

import (
	"context"
	"slices"
	"sync"
	"testing"
	"time"

	"github.com/nexuscrm/shared/pkg/contract"
	sharedErrors "github.com/nexuscrm/shared/pkg/errors"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/delegationv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionstatev1"
	querysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
	recordsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	searchsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/searchv1"
	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/codes"
	"google.golang.org/grpc/status"
)

type cancelAfterFirstSearchEngine struct {
	cancel context.CancelFunc
	calls  int
}

func (e *cancelAfterFirstSearchEngine) SearchWithSpec(
	context.Context,
	queryspec.SearchSpec,
	*models.UserSession,
) ([]contract.Record, error) {
	e.calls++
	if e.calls == 1 {
		e.cancel()
	}
	return nil, nil
}

type failingSearchEngine struct {
	err error
}

func (e *failingSearchEngine) SearchWithSpec(
	context.Context,
	queryspec.SearchSpec,
	*models.UserSession,
) ([]contract.Record, error) {
	return nil, e.err
}

type successfulSearchEngine struct {
	specs []queryspec.SearchSpec
}

type parallelPlannedSearchEngine struct {
	mu          sync.Mutex
	active      int
	maxActive   int
	specs       map[string]queryspec.SearchSpec
	bothStarted chan struct{}
	release     chan struct{}
	startedOnce sync.Once
}

func (e *parallelPlannedSearchEngine) SearchWithSpec(
	_ context.Context,
	spec queryspec.SearchSpec,
	_ *models.UserSession,
) ([]contract.Record, error) {
	e.mu.Lock()
	e.active++
	if e.active > e.maxActive {
		e.maxActive = e.active
	}
	e.specs[spec.ObjectAPIName] = spec
	if e.active >= 2 {
		e.startedOnce.Do(func() { close(e.bothStarted) })
	}
	e.mu.Unlock()
	<-e.release
	e.mu.Lock()
	e.active--
	e.mu.Unlock()
	if spec.ObjectAPIName == "Workspace" {
		return []contract.Record{{"Id": "workspace-1", "Name": "Alpha"}}, nil
	}
	return []contract.Record{{"Id": "work-item-1", "DueDate": "2026-07-28"}}, nil
}

type searchScopeOrderEngineFunc func(
	context.Context,
	*models.UserSession,
	[]string,
	permissionstatev1.Version,
) ([]string, error)

func (f searchScopeOrderEngineFunc) ScopeOrder(
	ctx context.Context,
	viewer *models.UserSession,
	candidates []string,
	floor permissionstatev1.Version,
) ([]string, error) {
	return f(ctx, viewer, candidates, floor)
}

func passthroughSearchScopeOrderEngine() SearchScopeOrderEngine {
	return searchScopeOrderEngineFunc(func(
		_ context.Context,
		_ *models.UserSession,
		candidates []string,
		_ permissionstatev1.Version,
	) ([]string, error) {
		return append([]string(nil), candidates...), nil
	})
}

func (e *successfulSearchEngine) SearchWithSpec(
	_ context.Context,
	spec queryspec.SearchSpec,
	_ *models.UserSession,
) ([]contract.Record, error) {
	e.specs = append(e.specs, spec)
	switch spec.ObjectAPIName {
	case "Workspace":
		return []contract.Record{{"Id": "workspace-1", "Name": "Alpha"}}, nil
	case "WorkItem":
		return []contract.Record{{
			"Id": "work-item-1", "WorkspaceId": "workspace-1", "DueDate": "2026-07-28",
		}}, nil
	default:
		return nil, sharedErrors.NewMalformedRequestError("unexpected object")
	}
}

func TestRecordSearchProjectsSuccessfulResultsAndGlobalLimit(t *testing.T) {
	baseCtx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	engine := &successfulSearchEngine{}
	server, err := NewRecordSearchServer(
		engine, passthroughSearchScopeOrderEngine(),
	)
	if err != nil {
		t.Fatal(err)
	}
	request := searchsemantic.Request{
		ObjectIDs: []string{"Workspace", "WorkItem"}, Term: "proof",
		SearchFields: []string{"Name"}, SelectedFields: []string{"Id", "Name", "WorkspaceId", "DueDate"},
		Limit: 2,
	}
	wireRequest, err := mapping.ToWireSearchRequest(baseCtx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := server.SearchRecords(recordSearchTestContext(t, baseCtx), wireRequest)
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireSearchResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if len(response.Records) != 2 ||
		response.Records[0].ObjectID != "Workspace" || response.Records[0].RecordID != "workspace-1" ||
		response.Records[1].ObjectID != "WorkItem" || response.Records[1].RecordID != "work-item-1" {
		t.Fatalf("search response = %#v", response)
	}
	if response.Receipts.CapabilityVersion != searchsemantic.CapabilityName ||
		response.Receipts.CatalogRevision != 17 {
		t.Fatalf("search receipts = %#v", response.Receipts)
	}
	if len(engine.specs) != 2 || engine.specs[0].Limit != 2 || engine.specs[1].Limit != 1 {
		t.Fatalf("search engine page limits = %#v", engine.specs)
	}
	for _, spec := range engine.specs {
		if spec.Term != request.Term ||
			len(spec.SearchFields) != len(request.SearchFields) ||
			len(spec.ReturnFields) != len(request.SelectedFields) {
			t.Fatalf("search engine spec = %#v", spec)
		}
	}
}

func TestRecordSearchExecutesOneObjectScopedPlanWithBoundedParallelism(t *testing.T) {
	baseCtx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	engine := &parallelPlannedSearchEngine{
		specs:       make(map[string]queryspec.SearchSpec),
		bothStarted: make(chan struct{}), release: make(chan struct{}),
	}
	server, err := NewRecordSearchServer(engine, passthroughSearchScopeOrderEngine())
	if err != nil {
		t.Fatal(err)
	}
	request := searchsemantic.Request{
		Term: "proof", Limit: 2,
		ObjectPlans: []searchsemantic.ObjectPlan{
			{ObjectID: "Workspace", SearchFields: []string{"Name"}, SelectedFields: []string{"Id", "Name"}, Limit: 1},
			{ObjectID: "WorkItem", SearchFields: []string{"Subject"}, SelectedFields: []string{"Id", "DueDate"}, Limit: 1},
		},
	}
	wireRequest, err := mapping.ToWireSearchRequest(baseCtx, request)
	if err != nil {
		t.Fatal(err)
	}
	type result struct {
		response *runtimev1.SearchRecordsResponse
		err      error
	}
	completed := make(chan result, 1)
	serverCtx := recordSearchTestContext(t, baseCtx)
	go func() {
		response, searchErr := server.SearchRecords(serverCtx, wireRequest)
		completed <- result{response: response, err: searchErr}
	}()
	select {
	case <-engine.bothStarted:
	case <-baseCtx.Done():
		t.Fatal("object-scoped search plans did not overlap")
	}
	close(engine.release)
	got := <-completed
	if got.err != nil {
		t.Fatal(got.err)
	}
	response, err := mapping.FromWireSearchResponse(request, got.response)
	if err != nil {
		t.Fatal(err)
	}
	if len(response.Records) != 2 ||
		response.Records[0].ObjectID != "Workspace" ||
		response.Records[1].ObjectID != "WorkItem" {
		t.Fatalf("deterministic response = %#v", response.Records)
	}
	engine.mu.Lock()
	defer engine.mu.Unlock()
	if engine.maxActive != 2 {
		t.Fatalf("max active searches = %d, want 2", engine.maxActive)
	}
	workspace := engine.specs["Workspace"]
	workItem := engine.specs["WorkItem"]
	if !slices.Equal(workspace.SearchFields, []string{"Name"}) ||
		!slices.Equal(workspace.ReturnFields, []string{"Id", "Name"}) ||
		!slices.Equal(workItem.SearchFields, []string{"Subject"}) ||
		!slices.Equal(workItem.ReturnFields, []string{"Id", "DueDate"}) {
		t.Fatalf("object-scoped specs = %#v", engine.specs)
	}
}

func TestPlatformSearchRecordIntersectsEngineFieldsWithRequestedSelection(t *testing.T) {
	recordID := "lead-1"
	email := "proof@example.invalid"
	secret := "must-not-cross-boundary"
	record, err := platformSearchRecord(querysemantic.Record{
		ObjectID: "Lead",
		Fields: []querysemantic.Field{
			{Name: "Id", Value: querysemantic.Value{Kind: querysemantic.ValueString, String: &recordID}},
			{Name: "Email", Value: querysemantic.Value{Kind: querysemantic.ValueString, String: &email}},
			{Name: "Secret__c", Value: querysemantic.Value{Kind: querysemantic.ValueString, String: &secret}},
		},
	}, []string{"Email"})
	if err != nil {
		t.Fatal(err)
	}
	if record.RecordID != recordID || len(record.Fields) != 1 ||
		record.Fields[0].Name != "Email" || record.Fields[0].String == nil ||
		*record.Fields[0].String != email {
		t.Fatalf("search record = %#v", record)
	}
}

func TestRecordSearchDoesNotStartAnotherObjectAfterCancellation(t *testing.T) {
	baseCtx, cancel := context.WithTimeout(context.Background(), time.Second)
	engine := &cancelAfterFirstSearchEngine{cancel: cancel}
	server, err := NewRecordSearchServer(
		engine, passthroughSearchScopeOrderEngine(),
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := recordSearchTestContext(t, baseCtx)
	request, err := mapping.ToWireSearchRequest(baseCtx, searchsemantic.Request{
		ObjectIDs: []string{"Workspace", "WorkItem"}, Term: "proof",
		SearchFields: []string{"Name"}, SelectedFields: []string{"Id"}, Limit: 2,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.SearchRecords(ctx, request); status.Code(err) != codes.Canceled {
		t.Fatalf("SearchRecords() status = %v, want Canceled", err)
	}
	if engine.calls != 1 {
		t.Fatalf("SearchWithSpec() calls = %d, want 1", engine.calls)
	}
}

func TestRecordSearchMapsEngineMalformedRequestToInvalidArgument(t *testing.T) {
	baseCtx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	engine := &failingSearchEngine{
		err: sharedErrors.NewMalformedRequestError("invalid search plan"),
	}
	server, err := NewRecordSearchServer(
		engine, passthroughSearchScopeOrderEngine(),
	)
	if err != nil {
		t.Fatal(err)
	}
	request, err := mapping.ToWireSearchRequest(baseCtx, searchsemantic.Request{
		ObjectIDs: []string{"WorkItem"}, Term: "proof",
		SearchFields: []string{"Subject"}, SelectedFields: []string{"Id"}, Limit: 1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.SearchRecords(
		recordSearchTestContext(t, baseCtx), request,
	); status.Code(err) != codes.InvalidArgument {
		t.Fatalf("SearchRecords() status = %v, want InvalidArgument", err)
	}
}

func TestSearchScopeOrderIntersectsRequestBoundViewBeforeProjection(t *testing.T) {
	base, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	var gotCandidates []string
	var gotViewer *models.UserSession
	var gotFloor permissionstatev1.Version
	scopeOrder := searchScopeOrderEngineFunc(func(
		_ context.Context,
		viewer *models.UserSession,
		candidates []string,
		floor permissionstatev1.Version,
	) ([]string, error) {
		gotCandidates = append([]string(nil), candidates...)
		gotViewer = viewer
		gotFloor = floor
		return []string{"WorkItem", "Workspace"}, nil
	})
	server, err := NewRecordSearchServer(&successfulSearchEngine{}, scopeOrder)
	if err != nil {
		t.Fatal(err)
	}
	revision := uint64(17)
	request := searchsemantic.ScopeOrderRequest{
		ObjectIDs:          []string{"Workspace", "HiddenObject", "WorkItem"},
		MinCatalogRevision: &revision,
	}
	wireRequest, err := mapping.ToWireSearchScopeOrderRequest(base, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := server.ReadSearchScopeOrder(
		recordSearchTestContext(t, base), wireRequest,
	)
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireSearchScopeOrderResponse(
		request, wireResponse,
	)
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Equal(gotCandidates, []string{"Workspace", "WorkItem"}) ||
		gotViewer == nil || gotViewer.ID != "user-1" ||
		gotFloor != effectivePermissionVersion(t, 7) {
		t.Fatalf(
			"projection authority candidates/viewer/floor = %v/%#v/%#v",
			gotCandidates, gotViewer, gotFloor,
		)
	}
	if !slices.Equal(response.ObjectIDs, []string{"WorkItem", "Workspace"}) ||
		response.Receipts.CatalogRevision != 17 ||
		response.Receipts.SecurityEpoch != 3 {
		t.Fatalf("scope-order response = %#v", response)
	}
}

func TestSearchScopeOrderRejectsUnservedCatalogRevisionBeforeProjection(
	t *testing.T,
) {
	base, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	called := false
	server, err := NewRecordSearchServer(
		&successfulSearchEngine{},
		searchScopeOrderEngineFunc(func(
			context.Context,
			*models.UserSession,
			[]string,
			permissionstatev1.Version,
		) ([]string, error) {
			called = true
			return nil, nil
		}),
	)
	if err != nil {
		t.Fatal(err)
	}
	required := uint64(18)
	request := searchsemantic.ScopeOrderRequest{
		ObjectIDs:          []string{"Workspace"},
		MinCatalogRevision: &required,
	}
	wireRequest, err := mapping.ToWireSearchScopeOrderRequest(base, request)
	if err != nil {
		t.Fatal(err)
	}
	_, err = server.ReadSearchScopeOrder(
		recordSearchTestContext(t, base), wireRequest,
	)
	if status.Code(err) != codes.FailedPrecondition || called {
		t.Fatalf(
			"unserved Catalog revision = %v, projection called = %v",
			err, called,
		)
	}
}

func TestSearchScopeOrderPropagatesCallerCancellation(t *testing.T) {
	base, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	started := make(chan struct{})
	projectionCanceled := make(chan struct{})
	server, err := NewRecordSearchServer(
		&successfulSearchEngine{},
		searchScopeOrderEngineFunc(func(
			ctx context.Context,
			_ *models.UserSession,
			_ []string,
			_ permissionstatev1.Version,
		) ([]string, error) {
			close(started)
			<-ctx.Done()
			close(projectionCanceled)
			return nil, ctx.Err()
		}),
	)
	if err != nil {
		t.Fatal(err)
	}
	request, err := mapping.ToWireSearchScopeOrderRequest(
		base,
		searchsemantic.ScopeOrderRequest{ObjectIDs: []string{"Workspace"}},
	)
	if err != nil {
		t.Fatal(err)
	}
	requestCtx, cancelRequest := context.WithCancel(
		recordSearchTestContext(t, base),
	)
	result := make(chan error, 1)
	go func() {
		_, invokeErr := server.ReadSearchScopeOrder(requestCtx, request)
		result <- invokeErr
	}()
	select {
	case <-started:
	case <-time.After(time.Second):
		cancelRequest()
		t.Fatal("scope-order projection did not start")
	}
	cancelRequest()
	select {
	case err := <-result:
		if status.Code(err) != codes.Canceled {
			t.Fatalf("canceled scope-order RPC = %v, want Canceled", err)
		}
	case <-time.After(time.Second):
		t.Fatal("canceled scope-order RPC did not return")
	}
	select {
	case <-projectionCanceled:
	case <-time.After(time.Second):
		t.Fatal("caller cancellation did not reach scope-order projection")
	}
}

func recordSearchTestContext(t *testing.T, baseCtx context.Context) context.Context {
	t.Helper()
	ctx := context.WithValue(baseCtx, runtimeExecutionScopeContextKey{}, RuntimeExecutionScope{
		view:              queryResponseTestView(t),
		orgID:             testOrgID,
		appInstallationID: testInstallationID,
		workloadID:        testWorkloadID,
		capability:        searchsemantic.CapabilityName,
		receipts: recordsemantic.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: searchsemantic.CapabilityName,
			CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 3,
		},
	})
	trusted := newTrustedCall(delegationv1.Claims{
		DelegationID: "delegation-1", OrgID: testOrgID, PrincipalID: "user-1", SessionID: "session-1",
		CallerWorkloadID: testWorkloadID, AppInstallationID: testInstallationID,
		RuntimeAudience: "runtime-test", Capabilities: []string{searchsemantic.CapabilityName},
		AuthEpoch: 2, SecurityEpoch: 3,
	}, models.UserSession{ID: "user-1"})
	trusted.permissionFloor = effectivePermissionVersion(t, 7)
	ctx = context.WithValue(ctx, trustedCallContextKey{}, trusted)
	return ctx
}