package runtimeapi

import (
	"context"
	"testing"
	"time"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	flowcommandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv1"
	recordsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type commandCancellationTestEngine struct {
	cancel context.CancelFunc
	t      *testing.T
}

func (e commandCancellationTestEngine) Accept(ctx context.Context, _ persistence.RuntimePlatformCommandAcceptance) (persistence.RuntimePlatformCommandRow, bool, error) {
	if e.cancel != nil {
		e.cancel()
		if ctx.Err() == nil {
			e.t.Fatal("cancel function did not cancel command request context")
		}
		return persistence.RuntimePlatformCommandRow{}, false, ctx.Err()
	}
	return persistence.RuntimePlatformCommandRow{}, false, context.Canceled
}

func TestCommandAcceptanceCancellationClassificationUsesRequestAuthority(t *testing.T) {
	for _, test := range []struct {
		name       string
		cancelCall bool
		want       codes.Code
	}{
		{name: "request canceled", cancelCall: true, want: codes.Canceled},
		{name: "owner returned cancellation", want: codes.Unavailable},
	} {
		t.Run(test.name, func(t *testing.T) {
			baseCtx, cancel := context.WithTimeout(context.Background(), time.Second)
			defer cancel()
			ctx := context.WithValue(baseCtx, runtimeExecutionScopeContextKey{}, RuntimeExecutionScope{
				view: commandAcceptanceTestView(t), orgID: testOrgID,
				appInstallationID: testInstallationID, workloadID: testWorkloadID,
				capability: commandsemantic.CapabilityName,
				receipts: recordsemantic.Receipts{
					ProtocolVersion: "1.0", CapabilityVersion: commandsemantic.CapabilityName,
					CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 3,
				},
			})
			ctx = context.WithValue(ctx, trustedWorkloadCallContextKey{}, TrustedWorkloadCall{
				orgID: testOrgID, callerWorkloadID: testWorkloadID,
				capability: commandsemantic.CapabilityName, securityEpoch: 3,
			})
			payload := []byte(`{"value":"test"}`)
			request, err := mapping.ToWireAcceptCommandRequest(ctx, commandsemantic.Request{
				CommandID: "command-1", IdempotencyKey: "request-1", CommandKind: "integration.ingress",
				PayloadFormatVersion: 1, Payload: payload,
				PayloadDigest: commandsemantic.ComputePayloadDigest(payload),
			})
			if err != nil {
				t.Fatal(err)
			}
			var cancelCall context.CancelFunc
			if test.cancelCall {
				cancelCall = cancel
			}
			server, err := NewCommandAcceptanceServer(commandCancellationTestEngine{cancel: cancelCall, t: t})
			if err != nil {
				t.Fatal(err)
			}
			if _, err := server.AcceptCommand(ctx, request); status.Code(err) != test.want {
				t.Fatalf("command status = %v, request context = %v, want %s", err, ctx.Err(), test.want)
			}
		})
	}
}

type commandCountingEngine struct {
	calls int
	last  persistence.RuntimePlatformCommandAcceptance
}

func (e *commandCountingEngine) Accept(_ context.Context, input persistence.RuntimePlatformCommandAcceptance) (persistence.RuntimePlatformCommandRow, bool, error) {
	e.calls++
	e.last = input
	return persistence.RuntimePlatformCommandRow{}, false, nil
}

func TestCommandAcceptanceRejectsUnregisteredKindAndVersionBeforePersistence(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	ctx = context.WithValue(ctx, runtimeExecutionScopeContextKey{}, RuntimeExecutionScope{
		view: commandAcceptanceTestView(t), orgID: testOrgID,
		appInstallationID: testInstallationID, workloadID: testWorkloadID,
		capability: commandsemantic.CapabilityName,
		receipts: recordsemantic.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: commandsemantic.CapabilityName,
			CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 3,
		},
	})
	ctx = context.WithValue(ctx, trustedWorkloadCallContextKey{}, TrustedWorkloadCall{
		orgID: testOrgID, callerWorkloadID: testWorkloadID,
		capability: commandsemantic.CapabilityName, securityEpoch: 3,
	})
	engine := &commandCountingEngine{}
	server, err := NewCommandAcceptanceServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	for _, request := range []commandsemantic.Request{
		validCommandRequest("other.effect", 1),
		validCommandRequest("integration.ingress", 2),
	} {
		wire, err := mapping.ToWireAcceptCommandRequest(ctx, request)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := server.AcceptCommand(ctx, wire); status.Code(err) != codes.InvalidArgument {
			t.Fatalf("unregistered command status = %v, want InvalidArgument", err)
		}
	}
	if engine.calls != 0 {
		t.Fatalf("durable engine calls = %d, want 0", engine.calls)
	}
}

func TestCommandAcceptanceKeepsOrdinaryCommandsHandlerOwned(t *testing.T) {
	view := commandAcceptanceTestView(t)
	scope := RuntimeExecutionScope{
		view: view, orgID: testOrgID,
		appInstallationID: testInstallationID, workloadID: testWorkloadID,
		capability: commandsemantic.CapabilityName,
		receipts: recordsemantic.Receipts{
			CatalogRevision: 17,
			SecurityEpoch:   3,
		},
	}
	request := validCommandRequest("integration.ingress", 1)
	if !commandRequestHasExecutionAuthority(t.Context(), scope, request) {
		t.Fatal("handler-owned ordinary command lost its published kind/version authority")
	}
	consumerContext := context.WithValue(
		t.Context(),
		commandConsumerAdmissionContextKey{},
		catalogsemantic.IntegrationBindingView{},
	)
	consumerContext = context.WithValue(
		consumerContext,
		runtimeExecutionScopeContextKey{},
		scope,
	)
	if commandRequestHasExecutionAuthority(consumerContext, scope, request) {
		t.Fatal("consumer-only admission accepted an ordinary command without exact binding grammar")
	}
}

func TestCommandAcceptanceRequiresExactFlowAuthorityBeforePersistence(t *testing.T) {
	exactPayload, err := flowcommandsemantic.Encode(
		flowcommandsemantic.Payload{
			DefinitionRef:     "flow.integration-ingress",
			DefinitionVersion: "flow-version-1",
			CanonicalInputs:   []byte(`{"recordId":"record-1"}`),
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	wrongDefinitionPayload, err := flowcommandsemantic.Encode(
		flowcommandsemantic.Payload{
			DefinitionRef:     "flow.other",
			DefinitionVersion: "flow-version-1",
			CanonicalInputs:   []byte(`{"recordId":"record-1"}`),
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	for _, test := range []struct {
		name            string
		bindingWorkload string
		executionBound  bool
		payload         []byte
		wantCode        codes.Code
		wantCalls       int
	}{
		{
			name: "exact workload and definition", bindingWorkload: testWorkloadID,
			executionBound: true, payload: exactPayload,
			wantCode: codes.OK, wantCalls: 1,
		},
		{
			name:    "N-1 registration has no definition authority",
			payload: exactPayload, wantCode: codes.InvalidArgument,
		},
		{
			name: "different workload", bindingWorkload: "other-workload",
			executionBound: true, payload: exactPayload,
			wantCode: codes.InvalidArgument,
		},
		{
			name: "different definition", bindingWorkload: testWorkloadID,
			executionBound: true, payload: wrongDefinitionPayload,
			wantCode: codes.InvalidArgument,
		},
		{
			name: "invalid Flow command payload", bindingWorkload: testWorkloadID,
			executionBound: true, payload: []byte(`{}`),
			wantCode: codes.InvalidArgument,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			baseCtx, cancel := context.WithTimeout(t.Context(), time.Second)
			defer cancel()
			var bindings []catalogsnapshot.CommandExecutionBinding
			if test.executionBound {
				bindings = []catalogsnapshot.CommandExecutionBinding{{
					AuthorizedWorkloadID: test.bindingWorkload,
					DefinitionRef:        "flow.integration-ingress",
					DefinitionVersion:    "flow-version-1",
					ContextNamespace:     "connector.test",
				}}
			}
			view := commandFlowAcceptanceTestView(t, bindings)
			ctx := context.WithValue(
				baseCtx,
				runtimeExecutionScopeContextKey{},
				RuntimeExecutionScope{
					view: view, orgID: testOrgID,
					appInstallationID: testInstallationID,
					workloadID:        testWorkloadID,
					capability:        commandsemantic.CapabilityName,
					receipts: recordsemantic.Receipts{
						ProtocolVersion:   "1.0",
						CapabilityVersion: commandsemantic.CapabilityName,
						CatalogRevision:   17, SchemaGeneration: 8,
						EngineGeneration: 9, SecurityEpoch: 3,
					},
				},
			)
			ctx = context.WithValue(
				ctx,
				trustedWorkloadCallContextKey{},
				TrustedWorkloadCall{
					orgID: testOrgID, callerWorkloadID: testWorkloadID,
					capability:    commandsemantic.CapabilityName,
					securityEpoch: 3,
				},
			)
			request := commandsemantic.Request{
				CommandID: "command-flow-1", IdempotencyKey: "request-flow-1",
				CommandKind:          flowcommandsemantic.CommandKind,
				PayloadFormatVersion: flowcommandsemantic.PayloadFormatVersion,
				Payload:              test.payload,
				PayloadDigest: commandsemantic.ComputePayloadDigest(
					test.payload,
				),
			}
			wire, err := mapping.ToWireAcceptCommandRequest(ctx, request)
			if err != nil {
				t.Fatal(err)
			}
			engine := &commandCountingEngine{}
			server, err := NewCommandAcceptanceServer(engine)
			if err != nil {
				t.Fatal(err)
			}
			_, err = server.AcceptCommand(ctx, wire)
			if code := status.Code(err); code != test.wantCode {
				t.Fatalf("AcceptCommand status = %s, want %s: %v",
					code, test.wantCode, err)
			}
			if engine.calls != test.wantCalls {
				t.Fatalf("durable engine calls = %d, want %d",
					engine.calls, test.wantCalls)
			}
		})
	}
}

func TestCommandAcceptancePersistsExactRecoverableExecutionAuthority(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	view := commandAcceptanceTestView(t)
	ctx = context.WithValue(ctx, runtimeExecutionScopeContextKey{}, RuntimeExecutionScope{
		view: view, orgID: testOrgID,
		appInstallationID: testInstallationID, workloadID: testWorkloadID,
		capability: commandsemantic.CapabilityName,
		receipts: recordsemantic.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: commandsemantic.CapabilityName,
			CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 3,
		},
	})
	ctx = context.WithValue(ctx, trustedWorkloadCallContextKey{}, TrustedWorkloadCall{
		orgID: testOrgID, callerWorkloadID: testWorkloadID,
		capability: commandsemantic.CapabilityName, securityEpoch: 3,
	})
	engine := &commandCountingEngine{}
	server, err := NewCommandAcceptanceServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	wire, err := mapping.ToWireAcceptCommandRequest(ctx, validCommandRequest("integration.ingress", 1))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.AcceptCommand(ctx, wire); err != nil {
		t.Fatal(err)
	}
	if engine.calls != 1 ||
		engine.last.OrgID != testOrgID ||
		engine.last.AppInstallationID != testInstallationID ||
		engine.last.CallerWorkloadID != testWorkloadID ||
		engine.last.RuntimeCapabilityID != commandsemantic.CapabilityName {
		t.Fatalf("accepted execution authority = %#v", engine.last)
	}
	binding, err := catalogsnapshot.OpenRuntimeWorkBinding(engine.last.MetadataBinding)
	if err != nil {
		t.Fatalf("accepted metadata binding: %v", err)
	}
	if err := binding.VerifyView(view, catalogsnapshot.ProjectionCommandAcceptanceV1); err != nil {
		t.Fatalf("accepted metadata binding does not reopen exact command view: %v", err)
	}
	expectedExecutionScopeDigest, err := (persistence.RuntimeWorkScope{
		OrgID:               testOrgID,
		AppInstallationID:   testInstallationID,
		RuntimeCapabilityID: commandsemantic.CapabilityName,
	}).Digest()
	if err != nil {
		t.Fatal(err)
	}
	if engine.last.ExecutionScopeDigest != expectedExecutionScopeDigest {
		t.Fatalf("execution scope digest = %q, want %q",
			engine.last.ExecutionScopeDigest, expectedExecutionScopeDigest)
	}
}

func validCommandRequest(kind string, version uint32) commandsemantic.Request {
	payload := []byte(`{"value":"test"}`)
	return commandsemantic.Request{
		CommandID: "command-1", IdempotencyKey: "request-1", CommandKind: kind,
		PayloadFormatVersion: version, Payload: payload,
		PayloadDigest: commandsemantic.ComputePayloadDigest(payload),
	}
}

func commandAcceptanceTestView(t *testing.T) *catalogsnapshot.RuntimeMetadataView {
	t.Helper()
	_, viewArtifact := commandAcceptanceTestArtifacts(t)
	view, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	return view
}

func commandFlowAcceptanceTestView(
	t *testing.T,
	executionBindings []catalogsnapshot.CommandExecutionBinding,
) *catalogsnapshot.RuntimeMetadataView {
	return commandFlowAcceptanceTestViewVersion(
		t,
		flowcommandsemantic.PayloadFormatVersion,
		executionBindings,
	)
}

func commandFlowAcceptanceTestViewVersion(
	t *testing.T,
	payloadFormatVersion uint32,
	executionBindings []catalogsnapshot.CommandExecutionBinding,
) *catalogsnapshot.RuntimeMetadataView {
	t.Helper()
	revision := catalogsnapshot.CatalogRevision(17)
	componentID := "component.command.runtime-flow"
	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-work-item-id", APIName: constants.FieldID,
						Label: "ID", Type: constants.FieldTypeString,
						PhysicalName: "id", IsSystem: true,
					}},
				},
			}},
			CommandRegistrations: []catalogsnapshot.CommandRegistrationInput{{
				CatalogRevision: revision, ComponentID: componentID,
				DefinitionVersion: "1",
				ContentDigest: catalogsnapshot.ComputeDigest(
					[]byte("runtime.flow.execute/test"),
				),
				CommandKind:          flowcommandsemantic.CommandKind,
				PayloadFormatVersion: payloadFormatVersion,
				ExecutionBindings:    executionBindings,
			}},
			Projections: []catalogsnapshot.RuntimeProjectionInput{{
				CatalogRevision:        revision,
				AppInstallationID:      testInstallationID,
				CapabilityID:           commandsemantic.CapabilityName,
				Profile:                catalogsnapshot.ProjectionCommandAcceptanceV1,
				DefinitionComponentIDs: []string{componentID},
			}},
		},
	)
	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:      commandsemantic.CapabilityName,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	view, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	if err := view.VerifySource(snapshot); err != nil {
		t.Fatal(err)
	}
	return view
}

func commandAcceptanceTestArtifacts(t *testing.T) (catalogsnapshot.Artifact, catalogsnapshot.Artifact) {
	t.Helper()
	revision := catalogsnapshot.CatalogRevision(17)
	componentID := "component.command.integration-ingress"
	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-work-item-id", APIName: constants.FieldID, Label: "ID",
					Type: constants.FieldTypeString, PhysicalName: "id", IsSystem: true,
				}},
			},
		}},
		CommandRegistrations: []catalogsnapshot.CommandRegistrationInput{{
			CatalogRevision: revision, ComponentID: componentID, DefinitionVersion: "1",
			ContentDigest: catalogsnapshot.ComputeDigest([]byte("integration.ingress/v1")),
			CommandKind:   "integration.ingress", PayloadFormatVersion: 1,
		}},
		Projections: []catalogsnapshot.RuntimeProjectionInput{{
			CatalogRevision: revision, AppInstallationID: testInstallationID,
			CapabilityID:           commandsemantic.CapabilityName,
			Profile:                catalogsnapshot.ProjectionCommandAcceptanceV1,
			DefinitionComponentIDs: []string{componentID},
		}},
	})
	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: commandsemantic.CapabilityName,
	})
	if err != nil {
		t.Fatal(err)
	}
	return snapshotArtifact, viewArtifact
}