package runtimeapi

import (
	"bytes"
	"context"
	"encoding/json"
	"strconv"
	"testing"
	"time"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/delegationv1"
	recordsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	reportsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/reportv1"
	"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/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
)

type reportDefinitionTestEngine struct {
	calls      int
	definition *models.ReportDefinition
	user       *models.UserSession
	result     *models.ReportResult
	err        error
}

func (e *reportDefinitionTestEngine) RunReportDefinitionPage(
	_ context.Context,
	definition *models.ReportDefinition,
	user *models.UserSession,
	_ map[string]interface{},
	_, _ int,
) (*models.ReportResult, error) {
	e.calls++
	e.definition = definition
	e.user = user
	if e.err != nil {
		return nil, e.err
	}
	if e.result != nil {
		return e.result, nil
	}
	return &models.ReportResult{
		ReportMetadata: *definition,
		Rows:           []models.SObject{{"Id": "row-1", "Subject": "proof"}},
	}, nil
}

func TestReportReadExecutesOnlyRequestBoundImmutableDefinition(t *testing.T) {
	engine := &reportDefinitionTestEngine{}
	server, err := NewReportReadServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	ctx := reportReadTestContext(t)
	request, err := mapping.ToWireReadReportRequest(ctx, reportsemantic.Request{
		ReportID: "report-open-work", CanonicalFilters: []byte(`{}`), Limit: 10,
	})
	if err != nil {
		t.Fatal(err)
	}
	response, err := server.ReadReport(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	semantic, err := mapping.FromWireReadReportResponse(response)
	if err != nil {
		t.Fatal(err)
	}
	if engine.calls != 1 || engine.definition == nil ||
		engine.definition.ID != "report-open-work" || engine.definition.Name != "Open work" ||
		engine.user == nil || engine.user.ID != "user-1" {
		t.Fatalf("engine call = %d, definition %#v, user %#v", engine.calls, engine.definition, engine.user)
	}
	if string(semantic.CanonicalResult) != `{"aggregates":[],"columns":["Id","Subject"],"definition":{"aggregates":[],"columns":["Id","Subject"],"displayMode":"TABULAR","filters":[],"grouping":[],"scope":"","sourceObject":"WorkItem","sourceRef":""},"definitionDigest":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","definitionVersion":"1","displayMode":"TABULAR","grouping":[],"name":"Open work","reportId":"report-open-work","rows":[{"Id":"row-1","Subject":"proof"}],"schema":"nexuscrm.report-result/v1","totalCount":0}` {
		t.Fatalf("CanonicalResult = %s", semantic.CanonicalResult)
	}
}

func TestMarshalReportResultCarriesExactImmutableDefinition(t *testing.T) {
	selected := &models.ReportDefinition{
		ID: "report-pipeline", Name: "Pipeline", SourceObject: "Opportunity",
		SourceRef: "OpportunityList", DisplayMode: "SUMMARY",
		FilterLogic: "1 AND 2", Scope: "myTeam",
		Columns: []string{"Name", "Amount"},
		Filters: []models.ReportFilter{
			{Column: "StageName", Operator: "equals", Value: "Open"},
			{Column: "Amount", Operator: "greaterThan", Value: "0"},
		},
		Grouping:   []string{"StageName"},
		Aggregates: []models.ReportAggregate{{Field: "Amount", Type: "SUM"}},
		DateFilter: &models.ReportDateFilter{
			Column: "CloseDate", DurationValue: "THIS_QUARTER",
		},
	}
	engineMetadata := *selected
	engineMetadata.Columns = []string{"Name", "Amount__physical"}
	engineMetadata.Grouping = []string{"StageName__physical"}
	result := &models.ReportResult{
		ReportMetadata:       engineMetadata,
		Rows:                 []models.SObject{{"Name": "Deal", "Amount__physical": 42, "__primary_id__": "opp-1"}},
		TotalCount:           1,
		AggregateDedupColumn: map[string]string{"Amount": "__primary_id__"},
	}
	payload, err := marshalReportResultV1(
		result, selected, "7",
		"sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
	)
	if err != nil {
		t.Fatal(err)
	}
	var document reportResultDocumentV1
	if err := json.Unmarshal(payload, &document); err != nil {
		t.Fatal(err)
	}
	if document.Definition.SourceObject != "Opportunity" ||
		document.Definition.SourceRef != "OpportunityList" ||
		document.Definition.Scope != "myTeam" ||
		len(document.Definition.Filters) != 2 ||
		len(document.Definition.Aggregates) != 1 ||
		document.Definition.Aggregates[0].Field != "Amount" ||
		document.Definition.Aggregates[0].Type != "SUM" ||
		document.Definition.DateFilter == nil ||
		document.Definition.DateFilter.DurationValue != "THIS_QUARTER" {
		t.Fatalf("immutable definition = %#v", document.Definition)
	}
	if len(document.Columns) != 2 || document.Columns[1] != "Amount__physical" ||
		len(document.Grouping) != 1 || document.Grouping[0] != "StageName__physical" {
		t.Fatalf("execution projection = %#v", document)
	}
	if document.AggregateDedupColumn["Amount"] != "__primary_id__" {
		t.Fatalf("aggregate dedup receipt = %#v", document.AggregateDedupColumn)
	}
}

func TestReportReadDoesNotLetEngineMutateSelectedDefinitionReceipt(t *testing.T) {
	engine := &reportDefinitionTestEngine{result: &models.ReportResult{
		ReportMetadata: *reportReadTestDefinition(),
		Rows:           []models.SObject{},
	}}
	server, err := NewReportReadServer(&mutatingReportDefinitionEngine{wrapped: engine})
	if err != nil {
		t.Fatal(err)
	}
	ctx := reportReadTestContext(t)
	request, err := mapping.ToWireReadReportRequest(ctx, reportsemantic.Request{
		ReportID: "report-open-work", CanonicalFilters: []byte(`{}`), Limit: 10,
	})
	if err != nil {
		t.Fatal(err)
	}
	response, err := server.ReadReport(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	semantic, err := mapping.FromWireReadReportResponse(response)
	if err != nil {
		t.Fatal(err)
	}
	var document reportResultDocumentV1
	if err := json.Unmarshal(semantic.CanonicalResult, &document); err != nil {
		t.Fatal(err)
	}
	if document.Definition.SourceObject != "WorkItem" ||
		len(document.Definition.Columns) != 2 || document.Definition.Columns[0] != "Id" {
		t.Fatalf("selected definition was mutated through engine input: %#v", document.Definition)
	}
}

type mutatingReportDefinitionEngine struct {
	wrapped *reportDefinitionTestEngine
}

func (e *mutatingReportDefinitionEngine) RunReportDefinitionPage(
	ctx context.Context,
	definition *models.ReportDefinition,
	user *models.UserSession,
	filters map[string]interface{},
	limit int,
	offset int,
) (*models.ReportResult, error) {
	definition.SourceObject = "mutated"
	definition.Columns[0] = "mutated"
	return e.wrapped.RunReportDefinitionPage(ctx, reportReadTestDefinition(), user, filters, limit, offset)
}

func TestReportReadMapsEngineNotFoundToNotFound(t *testing.T) {
	engine := &reportDefinitionTestEngine{
		err: appErrors.NewNotFoundError("report source", "missing"),
	}
	server, err := NewReportReadServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	ctx := reportReadTestContext(t)
	request, err := mapping.ToWireReadReportRequest(ctx, reportsemantic.Request{
		ReportID: "report-open-work", CanonicalFilters: []byte(`{}`), Limit: 10,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.ReadReport(ctx, request); status.Code(err) != codes.NotFound {
		t.Fatalf("ReadReport() status = %v, want NotFound", err)
	}
	if engine.calls != 1 {
		t.Fatalf("engine calls = %d, want one", engine.calls)
	}
}

func TestReportReadRejectsLegacyProductResultBeforeWire(t *testing.T) {
	engine := &reportDefinitionTestEngine{result: &models.ReportResult{
		ReportMetadata: *reportReadTestDefinition(),
		FactMap:        map[string]interface{}{"T!T": map[string]interface{}{}},
	}}
	server, err := NewReportReadServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	ctx := reportReadTestContext(t)
	request, err := mapping.ToWireReadReportRequest(ctx, reportsemantic.Request{
		ReportID: "report-open-work", CanonicalFilters: []byte(`{}`), Limit: 10,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.ReadReport(ctx, request); status.Code(err) != codes.Internal {
		t.Fatalf("ReadReport status = %v, want Internal", err)
	}
}

func TestReportReadRejectsUnownedAggregateDedupReceipt(t *testing.T) {
	engine := &reportDefinitionTestEngine{result: &models.ReportResult{
		ReportMetadata:       *reportReadTestDefinition(),
		AggregateDedupColumn: map[string]string{"NotAnAggregate": "__primary_id__"},
	}}
	server, err := NewReportReadServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	ctx := reportReadTestContext(t)
	request, err := mapping.ToWireReadReportRequest(ctx, reportsemantic.Request{
		ReportID: "report-open-work", CanonicalFilters: []byte(`{}`), Limit: 10,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.ReadReport(ctx, request); status.Code(err) != codes.Internal {
		t.Fatalf("ReadReport status = %v, want Internal", err)
	}
}

func TestReportReadRejectsMissingOrBlankAggregateDedupIdentity(t *testing.T) {
	for _, test := range []struct {
		name string
		row  models.SObject
	}{
		{name: "missing", row: models.SObject{"Amount": 10}},
		{name: "blank", row: models.SObject{"Amount": 10, "__primary_id__": ""}},
	} {
		t.Run(test.name, func(t *testing.T) {
			definition := reportReadTestDefinition()
			definition.Aggregates = []models.ReportAggregate{{Field: "Amount", Type: "SUM"}}
			engine := &reportDefinitionTestEngine{result: &models.ReportResult{
				ReportMetadata:       *definition,
				Rows:                 []models.SObject{test.row},
				AggregateDedupColumn: map[string]string{"Amount": "__primary_id__"},
			}}
			server, err := NewReportReadServer(engine)
			if err != nil {
				t.Fatal(err)
			}
			ctx := reportReadTestContext(t)
			request, err := mapping.ToWireReadReportRequest(ctx, reportsemantic.Request{
				ReportID: "report-open-work", CanonicalFilters: []byte(`{}`), Limit: 10,
			})
			if err != nil {
				t.Fatal(err)
			}
			if _, err := server.ReadReport(ctx, request); status.Code(err) != codes.Internal {
				t.Fatalf("ReadReport status = %v, want Internal", err)
			}
		})
	}
}

func TestReportReadCarriesValidAggregateDedupIdentityAcrossRows(t *testing.T) {
	definition := reportReadTestDefinition()
	definition.Aggregates = []models.ReportAggregate{{Field: "Amount", Type: "SUM"}}
	engine := &reportDefinitionTestEngine{result: &models.ReportResult{
		ReportMetadata: *definition,
		Rows: []models.SObject{
			{"Amount": 10, "__primary_id__": "opp-1"},
			{"Amount": 10, "__primary_id__": "opp-1"},
			{"Amount": 5, "__primary_id__": "opp-2"},
		},
		AggregateDedupColumn: map[string]string{"Amount": "__primary_id__"},
	}}
	server, err := NewReportReadServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	ctx := reportReadTestContext(t)
	request, err := mapping.ToWireReadReportRequest(ctx, reportsemantic.Request{
		ReportID: "report-open-work", CanonicalFilters: []byte(`{}`), Limit: 10,
	})
	if err != nil {
		t.Fatal(err)
	}
	response, err := server.ReadReport(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	semantic, err := mapping.FromWireReadReportResponse(response)
	if err != nil {
		t.Fatal(err)
	}
	var document reportResultDocumentV1
	if err := json.Unmarshal(semantic.CanonicalResult, &document); err != nil {
		t.Fatal(err)
	}
	if document.AggregateDedupColumn["Amount"] != "__primary_id__" || len(document.Rows) != 3 {
		t.Fatalf("dedup receipt = %#v, rows = %#v", document.AggregateDedupColumn, document.Rows)
	}
}

func TestReportReadFailsClosedWhenDefinitionIsOutsideAppView(t *testing.T) {
	engine := &reportDefinitionTestEngine{}
	server, err := NewReportReadServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	ctx := reportReadTestContext(t)
	request, err := mapping.ToWireReadReportRequest(ctx, reportsemantic.Request{
		ReportID: "report-other-app", CanonicalFilters: []byte(`{}`), Limit: 10,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.ReadReport(ctx, request); status.Code(err) != codes.NotFound {
		t.Fatalf("ReadReport status = %v, want NotFound", err)
	}
	if engine.calls != 0 {
		t.Fatalf("engine calls = %d, want zero", engine.calls)
	}
}

func TestReportReadRejectsStaleRevisionBeforeEngineAccess(t *testing.T) {
	engine := &reportDefinitionTestEngine{}
	server, err := NewReportReadServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	ctx := reportReadTestContext(t)
	minimum := uint64(18)
	request, err := mapping.ToWireReadReportRequest(ctx, reportsemantic.Request{
		ReportID: "report-open-work", CanonicalFilters: []byte(`{}`), Limit: 10,
		MinCatalogRevision: &minimum,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.ReadReport(ctx, request); status.Code(err) != codes.FailedPrecondition {
		t.Fatalf("ReadReport status = %v, want FailedPrecondition", err)
	}
	if engine.calls != 0 {
		t.Fatalf("engine calls = %d, want zero", engine.calls)
	}
}

type cancelAwareReportEngine struct {
	entered  chan struct{}
	canceled chan error
}

func (e *cancelAwareReportEngine) RunReportDefinitionPage(
	ctx context.Context,
	_ *models.ReportDefinition,
	_ *models.UserSession,
	_ map[string]interface{},
	_, _ int,
) (*models.ReportResult, error) {
	close(e.entered)
	<-ctx.Done()
	e.canceled <- ctx.Err()
	return nil, ctx.Err()
}

func TestRuntimeAPIProcessReportUsesOneEndpointAndPropagatesDeadline(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t, 17, "io.nexuscrm.sfcrm.report.read.v1",
		reportsemantic.CapabilityName, catalogsnapshot.ProjectionReportExecutionV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	engine := &cancelAwareReportEngine{
		entered: make(chan struct{}), canceled: make(chan error, 1),
	}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		PlatformResourceClassAdmissionLimits: map[runtimegrpc.ResourceClass]int{
			runtimegrpc.ResourceClassHeavyWork: 1,
		},
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			reportsemantic.CapabilityName: time.Second,
		},
		ReportEngine: engine, WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: testSecurityAdmission(t), ServingAdmission: fixture.admission,
	}, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection, err := grpc.NewClient(
		"dns:///"+process.GRPCAddress(),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	ctx, cancel := context.WithTimeout(reportProtocolContext(context.Background()), 100*time.Millisecond)
	defer cancel()
	request, err := mapping.ToWireReadReportRequest(ctx, reportsemantic.Request{
		ReportID: "report-widget", CanonicalFilters: []byte(`{}`), Limit: 10,
	})
	if err != nil {
		t.Fatal(err)
	}
	response, callErr := runtimev1.NewReportReadServiceClient(connection).ReadReport(ctx, request)
	if response != nil || status.Code(callErr) != codes.DeadlineExceeded {
		t.Fatalf("deadline result=%#v error=%v", response, callErr)
	}
	select {
	case <-engine.entered:
	default:
		t.Fatal("report request did not cross the registered Runtime endpoint")
	}
	select {
	case adapterErr := <-engine.canceled:
		if adapterErr == nil {
			t.Fatal("report engine observed nil cancellation")
		}
	case <-time.After(time.Second):
		t.Fatal("deadline did not propagate into report engine")
	}
}

func reportProtocolContext(ctx context.Context) context.Context {
	ctx = metadata.AppendToOutgoingContext(ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(reportsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(reportsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, reportsemantic.CapabilityName,
	)
	return withTestDelegation(ctx, mustSignTestDelegation(testClaimsFor(time.Now(), reportsemantic.CapabilityName)))
}

func reportReadTestContext(t *testing.T) context.Context {
	t.Helper()
	base, cancel := context.WithTimeout(context.Background(), time.Second)
	t.Cleanup(cancel)
	view := reportReadTestView(t)
	ctx := context.WithValue(base, runtimeExecutionScopeContextKey{}, RuntimeExecutionScope{
		view: view,
		receipts: recordsemantic.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: reportsemantic.CapabilityName,
			CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 3,
		},
	})
	return context.WithValue(ctx, trustedCallContextKey{}, newTrustedCall(delegationv1.Claims{
		DelegationID: "delegation-1", OrgID: testOrgID, PrincipalID: "user-1", SessionID: "session-1",
		CallerWorkloadID: testWorkloadID, AppInstallationID: testInstallationID,
		RuntimeAudience: "runtime-test", Capabilities: []string{reportsemantic.CapabilityName},
		AuthEpoch: 2, SecurityEpoch: 3,
	}, models.UserSession{ID: "user-1"}))
}

func reportReadTestView(t *testing.T) *catalogsnapshot.RuntimeMetadataView {
	t.Helper()
	revision := catalogsnapshot.CatalogRevision(17)
	digest := catalogsnapshot.Digest("sha256:" + string(bytes.Repeat([]byte{'a'}, 64)))
	snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(catalogsnapshot.CompileInput{
		OrgID: testOrgID,
		Lineage: catalogsnapshot.Lineage{
			CatalogRevision: revision, SchemaGeneration: 8, EngineGeneration: 9,
		},
		Objects: []catalogsnapshot.ObjectInput{{
			CatalogRevision: revision,
			Metadata: &models.ObjectMetadata{
				ID: "object-work-item", APIName: "WorkItem", Label: "Work Item", PluralLabel: "Work Items",
				SharingModel: constants.SharingModelPublicReadWrite,
				Fields: []models.FieldMetadata{
					{ID: "field-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeString},
					{ID: "field-subject", APIName: "Subject", Label: "Subject", Type: constants.FieldTypeString},
				},
			},
		}},
		ReportDefinitions: []catalogsnapshot.ReportDefinitionInput{{
			CatalogRevision: revision, ComponentID: "component.report.open-work",
			DefinitionVersion: "1", ContentDigest: digest,
			Definition: reportReadTestDefinition(),
		}},
		Projections: []catalogsnapshot.RuntimeProjectionInput{{
			CatalogRevision: revision, AppInstallationID: testInstallationID,
			CapabilityID: reportsemantic.CapabilityName, Profile: catalogsnapshot.ProjectionReportExecutionV1,
			ObjectAPINames: []string{"WorkItem"}, DefinitionComponentIDs: []string{"component.report.open-work"},
		}},
	})
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	viewArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{
		AppInstallationID: testInstallationID, CapabilityID: reportsemantic.CapabilityName,
	})
	if err != nil {
		t.Fatal(err)
	}
	view, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	return view
}

func reportReadTestDefinition() *models.ReportDefinition {
	return &models.ReportDefinition{
		ID: "report-open-work", Name: "Open work", SourceObject: "WorkItem",
		DisplayMode: "TABULAR", Columns: []string{"Id", "Subject"},
	}
}