package runtimeapi

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	delegation "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/delegationv1"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/relatedcountsv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type relatedCountAccessStub struct {
	mu     sync.Mutex
	calls  int
	object *string
	ids    []string
	access map[string]contract.RecordAccess
	err    error
	view   *catalogsnapshot.RuntimeMetadataView
}

func (s *relatedCountAccessStub) EvaluateRecordReadAccess(ctx context.Context, object *string, ids []string, _ *models.UserSession) (map[string]contract.RecordAccess, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.calls++
	s.object = object
	s.ids = append([]string(nil), ids...)
	s.view, _ = runtimeRecordsMetadataViewFromContext(ctx)
	return s.access, s.err
}

func TestRelatedCountDeduplicatesParentReadAuthorityInput(t *testing.T) {
	access := &relatedCountAccessStub{access: map[string]contract.RecordAccess{"P-1": {Read: true}}}
	query := &relatedCountQueryStub{results: map[string]*models.QueryResult{
		"WorkItem": {TotalSize: 2},
		"Note":     {TotalSize: 1},
	}}
	server, err := NewRelatedCountServer(access, query, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	response, err := invokeRelatedCount(t, relatedCountTestContext(t), server, semantic.Request{Counts: []semantic.CountSpec{
		{ParentRecordID: "P-1", ChildObjectID: "WorkItem", RelationshipField: "WorkspaceId"},
		{ParentRecordID: "P-1", ChildObjectID: "Note", RelationshipField: "ParentId"},
	}})
	if err != nil || len(response.Results) != 2 {
		t.Fatalf("response=%#v error=%v", response, err)
	}
	if access.calls != 1 || len(access.ids) != 1 || access.ids[0] != "P-1" {
		t.Fatalf("parent read authority calls=%d ids=%v, want one unique parent", access.calls, access.ids)
	}
}

type relatedCountQueryStub struct {
	mu       sync.Mutex
	calls    []queryspec.CountSpec
	rowCalls int
	results  map[string]*models.QueryResult
	errors   map[string]error
	wait     bool
	panic    bool
	view     *catalogsnapshot.RuntimeMetadataView
}

func (s *relatedCountQueryStub) QueryWithSpec(ctx context.Context, spec queryspec.QuerySpec, _ *models.UserSession) (*models.QueryResult, error) {
	s.mu.Lock()
	s.rowCalls++
	result, err := s.results[spec.ObjectAPIName], s.errors[spec.ObjectAPIName]
	wait := s.wait
	shouldPanic := s.panic
	s.mu.Unlock()
	if shouldPanic {
		panic("query engine panic")
	}
	if wait {
		<-ctx.Done()
		return nil, ctx.Err()
	}
	return result, err
}

func (s *relatedCountQueryStub) CountWithSpec(ctx context.Context, spec queryspec.CountSpec, _ *models.UserSession) (int, error) {
	s.mu.Lock()
	s.calls = append(s.calls, spec)
	s.view, _ = runtimeRecordsMetadataViewFromContext(ctx)
	result, err := s.results[spec.ObjectAPIName], s.errors[spec.ObjectAPIName]
	wait := s.wait
	shouldPanic := s.panic
	s.mu.Unlock()
	if shouldPanic {
		panic("query engine panic")
	}
	if wait {
		<-ctx.Done()
		return 0, ctx.Err()
	}
	if err != nil || result == nil {
		if err != nil {
			return 0, err
		}
		return 0, errors.New("nil exact count result")
	}
	return result.TotalSize, nil
}

func (*relatedCountQueryStub) QueryHistory(context.Context, contract.HistoryQueryRequest, *models.UserSession) (*contract.HistoryQueryResult, error) {
	return nil, errors.New("unexpected QueryHistory call")
}

func relatedCountTestContext(t *testing.T) context.Context {
	t.Helper()
	base, cancel := context.WithTimeout(context.Background(), time.Second)
	t.Cleanup(cancel)
	trusted := newTrustedCall(delegation.Claims{
		DelegationID: "delegation-related-count", OrgID: "org-related-count", PrincipalID: "user-related-count",
		SessionID: "session-related-count", CallerWorkloadID: "appservice-related-count",
		AppInstallationID: "installation-related-count", RuntimeAudience: "runtime-related-count",
		Capabilities: []string{semantic.CapabilityName}, AuthEpoch: 1, SecurityEpoch: 1,
	}, models.UserSession{ID: "user-related-count"})
	ctx := context.WithValue(base, trustedCallContextKey{}, trusted)
	queryView := queryResponseTestView(t)
	ctx = context.WithValue(ctx, runtimeExecutionScopeContextKey{}, RuntimeExecutionScope{
		view: queryView,
		receipts: semantic.Receipts{
			CatalogRevision: 1, SchemaGeneration: 2, EngineGeneration: 3, SecurityEpoch: 1,
		},
	})
	return context.WithValue(ctx, runtimeRecordAccessMetadataViewContextKey{}, &catalogsnapshot.RuntimeMetadataView{})
}

func TestRelatedCountUsesRecordAccessSiblingOnlyForParentAuthority(t *testing.T) {
	access := &relatedCountAccessStub{access: map[string]contract.RecordAccess{"P-1": {Read: true}}}
	query := &relatedCountQueryStub{results: map[string]*models.QueryResult{"WorkItem": {TotalSize: 2}}}
	server, err := NewRelatedCountServer(access, query, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	ctx := relatedCountTestContext(t)
	primaryView, _ := RuntimeMetadataViewFromContext(ctx)
	accessView, _ := runtimeRecordAccessMetadataViewFromContext(ctx)
	if primaryView == accessView {
		t.Fatal("test requires distinct record-query and record-access views")
	}
	response, err := invokeRelatedCount(t, ctx, server, semantic.Request{Counts: []semantic.CountSpec{{
		ParentRecordID: "P-1", ChildObjectID: "WorkItem", RelationshipField: "WorkspaceId",
	}}})
	if err != nil || len(response.Results) != 1 || response.Results[0].Count != 2 {
		t.Fatalf("response=%#v error=%v", response, err)
	}
	if access.view != accessView {
		t.Fatalf("parent authority view = %p, want record-access sibling %p", access.view, accessView)
	}
	if query.view != primaryView {
		t.Fatalf("child count view = %p, want primary record-query view %p", query.view, primaryView)
	}
}

func invokeRelatedCount(t *testing.T, ctx context.Context, server *RelatedCountServer, request semantic.Request) (semantic.Response, error) {
	t.Helper()
	wireRequest, err := mapping.ToWireCountRelatedRecordsRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, wireErr := server.CountRelatedRecords(ctx, wireRequest)
	if wireErr != nil {
		return semantic.Response{}, mapping.FromRelatedCountStatusError(wireErr)
	}
	return mapping.FromWireCountRelatedRecordsResponse(request, wireResponse)
}

func TestRelatedCountPreservesOrderAndMasksParentVisibility(t *testing.T) {
	access := &relatedCountAccessStub{access: map[string]contract.RecordAccess{
		"parent-readable":      {Read: true},
		"parent-child-missing": {Read: true},
	}}
	query := &relatedCountQueryStub{
		results: map[string]*models.QueryResult{"WorkItem": {TotalSize: 3}},
		errors:  map[string]error{"MissingChild": appErrors.NewNotFoundError("Object", "MissingChild")},
	}
	server, err := NewRelatedCountServer(access, query, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	request := semantic.Request{Counts: []semantic.CountSpec{
		{ParentRecordID: "parent-readable", ChildObjectID: "WorkItem", RelationshipField: "WorkspaceId", MatchValue: "workspace-key"},
		{ParentRecordID: "parent-hidden", ChildObjectID: "NeverQueried", RelationshipField: "WorkspaceId"},
		{ParentRecordID: "parent-child-missing", ChildObjectID: "MissingChild", RelationshipField: "WorkspaceId"},
	}}
	response, err := invokeRelatedCount(t, relatedCountTestContext(t), server, request)
	if err != nil {
		t.Fatal(err)
	}
	if len(response.Results) != 3 || response.Results[0] != (semantic.Result{Index: 0, Outcome: semantic.CountOutcomeCounted, Count: 3}) ||
		response.Results[1] != (semantic.Result{Index: 1, Outcome: semantic.CountOutcomeCounted}) ||
		response.Results[2] != (semantic.Result{Index: 2, Outcome: semantic.CountOutcomeNotAvailable}) {
		t.Fatalf("results = %#v", response.Results)
	}
	if response.Receipts.CapabilityVersion != semantic.CapabilityName || response.Receipts.CatalogRevision != 1 {
		t.Fatalf("receipts = %#v", response.Receipts)
	}
	if access.calls != 1 || access.object != nil || len(access.ids) != 3 {
		t.Fatalf("parent access calls=%d object=%v ids=%v", access.calls, access.object, access.ids)
	}
	query.mu.Lock()
	defer query.mu.Unlock()
	if len(query.calls) != 2 {
		t.Fatalf("query calls = %#v, want only readable parents", query.calls)
	}
	for _, call := range query.calls {
		if call.ObjectAPIName == "WorkItem" && (call.Filter.Field != "WorkspaceId" || call.Filter.Value != "workspace-key") {
			t.Fatalf("query spec = %#v", call)
		}
	}
	if query.rowCalls != 0 {
		t.Fatalf("row query calls = %d, want exact count path only", query.rowCalls)
	}
}

func TestRelatedCountEmptyBatchDoesNotReachEngines(t *testing.T) {
	access := &relatedCountAccessStub{}
	query := &relatedCountQueryStub{}
	server, err := NewRelatedCountServer(access, query, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	response, err := invokeRelatedCount(t, relatedCountTestContext(t), server, semantic.Request{})
	if err != nil || len(response.Results) != 0 || access.calls != 0 || len(query.calls) != 0 {
		t.Fatalf("response=%#v error=%v access=%d queries=%d", response, err, access.calls, len(query.calls))
	}
}

func TestRelatedCountMasksParentObjectReadDenialAsCountedZero(t *testing.T) {
	access := &relatedCountAccessStub{err: appErrors.NewPermissionError("read", "Workspace")}
	query := &relatedCountQueryStub{}
	server, err := NewRelatedCountServer(access, query, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	response, err := invokeRelatedCount(t, relatedCountTestContext(t), server, semantic.Request{Counts: []semantic.CountSpec{
		{ParentRecordID: "P-1", ChildObjectID: "WorkItem", RelationshipField: "WorkspaceId"},
	}})
	if err != nil || len(response.Results) != 1 || response.Results[0] != (semantic.Result{Index: 0, Outcome: semantic.CountOutcomeCounted}) || len(query.calls) != 0 {
		t.Fatalf("response=%#v error=%v queries=%d", response, err, len(query.calls))
	}
}

func TestRelatedCountRejectsNilEngineResultAsInternal(t *testing.T) {
	access := &relatedCountAccessStub{access: map[string]contract.RecordAccess{"P-1": {Read: true}}}
	query := &relatedCountQueryStub{results: map[string]*models.QueryResult{"WorkItem": nil}}
	server, err := NewRelatedCountServer(access, query, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	_, err = invokeRelatedCount(t, relatedCountTestContext(t), server, semantic.Request{Counts: []semantic.CountSpec{
		{ParentRecordID: "P-1", ChildObjectID: "WorkItem", RelationshipField: "WorkspaceId"},
	}})
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureInternal {
		t.Fatalf("error = %#v", err)
	}
}

func TestRelatedCountContainsQueryEnginePanicAsInternal(t *testing.T) {
	access := &relatedCountAccessStub{access: map[string]contract.RecordAccess{"P-1": {Read: true}}}
	query := &relatedCountQueryStub{panic: true}
	server, err := NewRelatedCountServer(access, query, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	_, err = invokeRelatedCount(t, relatedCountTestContext(t), server, semantic.Request{Counts: []semantic.CountSpec{
		{ParentRecordID: "P-1", ChildObjectID: "WorkItem", RelationshipField: "WorkspaceId"},
	}})
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureInternal {
		t.Fatalf("error = %#v", err)
	}
}

func TestRelatedCountRequiresBoundMetadataBeforeEngineAccess(t *testing.T) {
	access := &relatedCountAccessStub{}
	query := &relatedCountQueryStub{}
	server, err := NewRelatedCountServer(access, query, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	ctx := relatedCountTestContext(t)
	trusted, _ := trustedCallFromContext(ctx)
	base, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	ctx = context.WithValue(base, trustedCallContextKey{}, trusted)
	_, err = invokeRelatedCount(t, ctx, server, semantic.Request{Counts: []semantic.CountSpec{{ParentRecordID: "P-1", ChildObjectID: "WorkItem", RelationshipField: "WorkspaceId"}}})
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureUnavailable || access.calls != 0 || len(query.calls) != 0 {
		t.Fatalf("error=%#v access=%d queries=%d", err, access.calls, len(query.calls))
	}
}

func TestRelatedCountRequiresRecordAccessSiblingBeforeEngineAccess(t *testing.T) {
	access := &relatedCountAccessStub{}
	query := &relatedCountQueryStub{}
	server, err := NewRelatedCountServer(access, query, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	ctx := relatedCountTestContext(t)
	ctx = context.WithValue(ctx, runtimeRecordAccessMetadataViewContextKey{}, (*catalogsnapshot.RuntimeMetadataView)(nil))
	_, err = invokeRelatedCount(t, ctx, server, semantic.Request{Counts: []semantic.CountSpec{{
		ParentRecordID: "P-1", ChildObjectID: "WorkItem", RelationshipField: "WorkspaceId",
	}}})
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureUnavailable || access.calls != 0 || len(query.calls) != 0 {
		t.Fatalf("error=%#v access=%d queries=%d", err, access.calls, len(query.calls))
	}
}

func TestRelatedCountCancellationAbortsWholeBatch(t *testing.T) {
	access := &relatedCountAccessStub{access: map[string]contract.RecordAccess{"P-1": {Read: true}}}
	query := &relatedCountQueryStub{wait: true}
	server, err := NewRelatedCountServer(access, query, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(relatedCountTestContext(t), 20*time.Millisecond)
	defer cancel()
	_, err = invokeRelatedCount(t, ctx, server, semantic.Request{Counts: []semantic.CountSpec{{ParentRecordID: "P-1", ChildObjectID: "WorkItem", RelationshipField: "WorkspaceId"}}})
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureDeadlineExceeded || status.Code(mapping.ToRelatedCountStatusError(err)) != codes.DeadlineExceeded {
		t.Fatalf("error = %#v", err)
	}
}