package runtimeapi

import (
	"context"
	"errors"
	"strings"
	"testing"
	"time"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	flowcommandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv1"
	flowcommandsemanticv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
	flowcommandsemanticv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
	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"
)

func TestCommandConsumerAdmissionRequiresCurrentExactAuthorityBeforePersistence(
	t *testing.T,
) {
	runtimeView := commandFlowAcceptanceTestView(
		t,
		[]catalogsnapshot.CommandExecutionBinding{{
			AuthorizedWorkloadID: testWorkloadID,
			DefinitionRef:        "flow.integration-ingress",
			DefinitionVersion:    "flow-version-1",
			ContextNamespace:     "connector.test",
		}},
	)
	current := commandConsumerIntegrationView(runtimeView)
	engine := &catalogConsumerViewTestEngine{integrationView: current}
	admission, err := NewCommandConsumerAdmission(
		t.Context(),
		CommandConsumerSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID: testWorkloadID, RuntimeMetadataView: runtimeView,
		},
		engine,
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := commandConsumerTestContext(t, testWorkloadID, 3)
	bound, err := admission.bind(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(engine.integrationScopes) != 1 {
		t.Fatalf("Integration view reads = %d, want 1", len(engine.integrationScopes))
	}
	requireCatalogConsumerScope(
		t,
		engine.integrationScopes[0],
		catalogsemantic.IntegrationBindingCapabilityName,
	)
	if admitted, err := (&ServingAdmission{}).bind(
		bound,
		commandsemantic.CapabilityName,
	); err != nil || admitted != bound {
		t.Fatalf("command consumer serving handoff = %v, %v", admitted, err)
	}

	payload, err := flowcommandsemantic.Encode(flowcommandsemantic.Payload{
		DefinitionRef:     "flow.integration-ingress",
		DefinitionVersion: "flow-version-1",
		CanonicalInputs:   []byte(`{"recordId":"record-1"}`),
	})
	if err != nil {
		t.Fatal(err)
	}
	request := commandsemantic.Request{
		CommandID: "command-flow-1", IdempotencyKey: "request-flow-1",
		CommandKind:          flowcommandsemantic.CommandKind,
		PayloadFormatVersion: flowcommandsemantic.PayloadFormatVersion,
		Payload:              payload,
		PayloadDigest:        commandsemantic.ComputePayloadDigest(payload),
	}
	requestContext, cancel := context.WithTimeout(bound, time.Second)
	defer cancel()
	wire, err := mapping.ToWireAcceptCommandRequest(requestContext, request)
	if err != nil {
		t.Fatal(err)
	}
	commandEngine := &commandCountingEngine{}
	server, err := NewCommandAcceptanceServer(commandEngine)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.AcceptCommand(requestContext, wire); err != nil {
		t.Fatal(err)
	}
	if commandEngine.calls != 1 {
		t.Fatalf("durable engine calls = %d, want 1", commandEngine.calls)
	}

	// The process-pinned authority is immutable. Mutating the reader after
	// assembly must neither trigger another Catalog read nor change request
	// admission; a stopped revision cutover assembles a new process.
	current.Bindings[0].RuntimeCommand = nil
	current.Lineage.ViewDigest =
		catalogsemantic.ComputeIntegrationBindingViewDigest(current)
	engine.integrationView = current
	bound, err = admission.bind(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(engine.integrationScopes) != 1 {
		t.Fatalf("repeated Integration view reads = %d, want 1 total", len(engine.integrationScopes))
	}
	removedContext, removedCancel := context.WithTimeout(bound, time.Second)
	defer removedCancel()
	commandEngine = &commandCountingEngine{}
	server, err = NewCommandAcceptanceServer(commandEngine)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.AcceptCommand(removedContext, wire); err != nil {
		t.Fatalf("pinned command admission = %v", err)
	}
	if commandEngine.calls != 1 {
		t.Fatalf(
			"pinned command durable persistence: %d calls, want 1",
			commandEngine.calls,
		)
	}
}

func TestCommandConsumerAdmissionAcceptsOnlyCatalogSelectedV2DependencyCommand(
	t *testing.T,
) {
	runtimeView := commandFlowAcceptanceTestViewVersion(
		t,
		flowcommandsemanticv2.PayloadFormatVersion,
		[]catalogsnapshot.CommandExecutionBinding{{
			AuthorizedWorkloadID: testWorkloadID,
			DefinitionRef:        "flow.integration-ingress",
			DefinitionVersion:    "flow-version-1",
			ContextNamespace:     "connector.test",
		}},
	)
	current := commandConsumerIntegrationView(runtimeView)
	current.Bindings[0].RuntimeCommand.PayloadFormatVersion =
		flowcommandsemanticv2.PayloadFormatVersion
	current.Lineage.ViewDigest =
		catalogsemantic.ComputeIntegrationBindingViewDigest(current)
	engine := &catalogConsumerViewTestEngine{integrationView: current}
	admission, err := NewCommandConsumerAdmission(
		t.Context(),
		CommandConsumerSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID: testWorkloadID, RuntimeMetadataView: runtimeView,
		},
		engine,
	)
	if err != nil {
		t.Fatal(err)
	}
	bound, err := admission.bind(commandConsumerTestContext(t, testWorkloadID, 3))
	if err != nil {
		t.Fatal(err)
	}
	payload, err := flowcommandsemanticv2.Encode(flowcommandsemanticv2.Payload{
		DefinitionRef:     "flow.integration-ingress",
		DefinitionVersion: "flow-version-1",
		CanonicalInputs:   []byte(`{"recordId":"record-1"}`),
		DependencySource: flowcommandsemanticv2.DependencySourceScope{
			ObjectKind: "contacts", SourceID: "remote-contact-1",
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	request := commandsemantic.Request{
		CommandID: "command-flow-v2", IdempotencyKey: "request-flow-v2",
		CommandKind:          flowcommandsemantic.CommandKind,
		PayloadFormatVersion: flowcommandsemanticv2.PayloadFormatVersion,
		Payload:              payload,
		PayloadDigest:        commandsemantic.ComputePayloadDigest(payload),
	}
	requestContext, cancel := context.WithTimeout(bound, time.Second)
	defer cancel()
	wire, err := mapping.ToWireAcceptCommandRequest(requestContext, request)
	if err != nil {
		t.Fatal(err)
	}
	commandEngine := &commandCountingEngine{}
	server, err := NewCommandAcceptanceServer(commandEngine)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.AcceptCommand(requestContext, wire); err != nil {
		t.Fatal(err)
	}
	if commandEngine.calls != 1 ||
		commandEngine.last.PayloadFormatVersion !=
			flowcommandsemanticv2.PayloadFormatVersion {
		t.Fatalf("V2 durable acceptance = %#v", commandEngine.last)
	}

	current.Bindings[0].RuntimeCommand.PayloadFormatVersion =
		flowcommandsemantic.PayloadFormatVersion
	current.Lineage.ViewDigest =
		catalogsemantic.ComputeIntegrationBindingViewDigest(current)
	engine.integrationView = current
	// A stopped Catalog cutover assembles a new admission and pins the newly
	// verified binding. It must not borrow the previous process authority.
	admission, err = NewCommandConsumerAdmission(
		t.Context(),
		CommandConsumerSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID: testWorkloadID, RuntimeMetadataView: runtimeView,
		},
		engine,
	)
	if err != nil {
		t.Fatal(err)
	}
	bound, err = admission.bind(commandConsumerTestContext(t, testWorkloadID, 3))
	if err != nil {
		t.Fatal(err)
	}
	mismatchContext, mismatchCancel := context.WithTimeout(bound, time.Second)
	defer mismatchCancel()
	commandEngine = &commandCountingEngine{}
	server, err = NewCommandAcceptanceServer(commandEngine)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.AcceptCommand(mismatchContext, wire); status.Code(err) !=
		codes.InvalidArgument {
		t.Fatalf("V2 against V1 Catalog binding status = %v, want InvalidArgument", err)
	}
	if commandEngine.calls != 0 {
		t.Fatalf("mismatched V2 command reached persistence: %d calls", commandEngine.calls)
	}
}

func TestCommandConsumerAdmissionAcceptsOnlyCatalogSelectedV3ApplicationCommand(
	t *testing.T,
) {
	runtimeView := commandFlowAcceptanceTestViewVersion(
		t,
		flowcommandsemanticv3.PayloadFormatVersion,
		[]catalogsnapshot.CommandExecutionBinding{{
			AuthorizedWorkloadID: testWorkloadID,
			DefinitionRef:        "flow.integration-ingress",
			DefinitionVersion:    "flow-version-1",
			ContextNamespace:     "connector.test",
		}},
	)
	current := commandConsumerIntegrationView(runtimeView)
	current.Bindings[0].RuntimeCommand.PayloadFormatVersion =
		flowcommandsemanticv3.PayloadFormatVersion
	current.Lineage.ViewDigest =
		catalogsemantic.ComputeIntegrationBindingViewDigest(current)
	engine := &catalogConsumerViewTestEngine{integrationView: current}
	admission, err := NewCommandConsumerAdmission(
		t.Context(),
		CommandConsumerSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID: testWorkloadID, RuntimeMetadataView: runtimeView,
		},
		engine,
	)
	if err != nil {
		t.Fatal(err)
	}
	bound, err := admission.bind(commandConsumerTestContext(t, testWorkloadID, 3))
	if err != nil {
		t.Fatal(err)
	}
	canonicalInputs := []byte(`{"recordId":"record-1"}`)
	inputDigest, err := flowcommandsemanticv3.ComputeInputDigest(
		canonicalInputs,
		flowcommandsemanticv3.SuppressExternalEffects,
	)
	if err != nil {
		t.Fatal(err)
	}
	payload, err := flowcommandsemanticv3.Encode(flowcommandsemanticv3.Payload{
		ExecutionRef: applicationresultsemantic.ExecutionRef{
			ExecutionID: "connector-execution-1", PackGenerationID: "pack-generation-1",
			ServingEpochDigest: "sha256:" + strings.Repeat("1", 64),
			DefinitionRef:      "flow.integration-ingress", DefinitionVersion: "flow-version-1",
			InputDigest: inputDigest,
		},
		CanonicalInputs:     canonicalInputs,
		ExternalEffectsMode: flowcommandsemanticv3.SuppressExternalEffects,
	})
	if err != nil {
		t.Fatal(err)
	}
	request := commandsemantic.Request{
		CommandID: "command-flow-v3", IdempotencyKey: "request-flow-v3",
		CommandKind:          flowcommandsemantic.CommandKind,
		PayloadFormatVersion: flowcommandsemanticv3.PayloadFormatVersion,
		Payload:              payload,
		PayloadDigest:        commandsemantic.ComputePayloadDigest(payload),
	}
	requestContext, cancel := context.WithTimeout(bound, time.Second)
	defer cancel()
	wire, err := mapping.ToWireAcceptCommandRequest(requestContext, request)
	if err != nil {
		t.Fatal(err)
	}
	commandEngine := &commandCountingEngine{}
	server, err := NewCommandAcceptanceServer(commandEngine)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.AcceptCommand(requestContext, wire); err != nil {
		t.Fatal(err)
	}
	if commandEngine.calls != 1 ||
		commandEngine.last.PayloadFormatVersion != flowcommandsemanticv3.PayloadFormatVersion {
		t.Fatalf("V3 durable acceptance = %#v", commandEngine.last)
	}

	current.Bindings[0].RuntimeCommand.PayloadFormatVersion =
		flowcommandsemanticv2.PayloadFormatVersion
	current.Lineage.ViewDigest =
		catalogsemantic.ComputeIntegrationBindingViewDigest(current)
	engine.integrationView = current
	admission, err = NewCommandConsumerAdmission(
		t.Context(),
		CommandConsumerSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID: testWorkloadID, RuntimeMetadataView: runtimeView,
		},
		engine,
	)
	if err != nil {
		t.Fatal(err)
	}
	bound, err = admission.bind(commandConsumerTestContext(t, testWorkloadID, 3))
	if err != nil {
		t.Fatal(err)
	}
	mismatchContext, mismatchCancel := context.WithTimeout(bound, time.Second)
	defer mismatchCancel()
	commandEngine = &commandCountingEngine{}
	server, err = NewCommandAcceptanceServer(commandEngine)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.AcceptCommand(mismatchContext, wire); status.Code(err) !=
		codes.InvalidArgument {
		t.Fatalf("V3 against V2 Catalog binding status = %v, want InvalidArgument", err)
	}
	if commandEngine.calls != 0 {
		t.Fatalf("mismatched V3 command reached persistence: %d calls", commandEngine.calls)
	}
}

func TestCommandConsumerAdmissionFailsClosedOnScopeAndLineage(t *testing.T) {
	runtimeView := commandFlowAcceptanceTestView(
		t,
		[]catalogsnapshot.CommandExecutionBinding{{
			AuthorizedWorkloadID: testWorkloadID,
			DefinitionRef:        "flow.integration-ingress",
			DefinitionVersion:    "flow-version-1",
			ContextNamespace:     "connector.test",
		}},
	)
	current := commandConsumerIntegrationView(runtimeView)
	for _, test := range []struct {
		name      string
		workload  string
		mutate    func(*catalogsemantic.IntegrationBindingView)
		prepare   bool
		want      recordsemantic.FailureCategory
		staleAxis recordsemantic.StaleAxis
	}{
		{
			name: "wrong workload", workload: "other-workload",
			want: recordsemantic.FailureForbidden,
		},
		{
			name: "mixed Catalog revision",
			mutate: func(view *catalogsemantic.IntegrationBindingView) {
				view.Lineage.CatalogRevision++
				view.Receipts.CatalogRevision++
				view.Lineage.ViewDigest =
					catalogsemantic.ComputeIntegrationBindingViewDigest(*view)
			},
			want:      recordsemantic.FailureStaleState,
			staleAxis: recordsemantic.StaleAxisCatalog,
			prepare:   true,
		},
		{
			name: "different source snapshot",
			mutate: func(view *catalogsemantic.IntegrationBindingView) {
				view.Lineage.SourceSnapshotDigest =
					"sha256:" + strings.Repeat("a", 64)
				view.Lineage.ViewDigest =
					catalogsemantic.ComputeIntegrationBindingViewDigest(*view)
			},
			want:      recordsemantic.FailureStaleState,
			staleAxis: recordsemantic.StaleAxisCatalog,
			prepare:   true,
		},
		{
			name: "stale SecurityEpoch",
			mutate: func(view *catalogsemantic.IntegrationBindingView) {
				view.Receipts.SecurityEpoch++
			},
			want:      recordsemantic.FailureStaleState,
			staleAxis: recordsemantic.StaleAxisSecurity,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			view := current
			if test.mutate != nil {
				test.mutate(&view)
			}
			admission, err := NewCommandConsumerAdmission(
				t.Context(),
				CommandConsumerSelection{
					OrgID: testOrgID, AppInstallationID: testInstallationID,
					WorkloadID:          testWorkloadID,
					RuntimeMetadataView: runtimeView,
				},
				&catalogConsumerViewTestEngine{integrationView: view},
			)
			if test.prepare {
				var failure *recordsemantic.Failure
				if !errors.As(err, &failure) ||
					failure.Category != test.want ||
					failure.StaleAxis != test.staleAxis {
					t.Fatalf("preparation error = %#v, want %s/%s", err, test.want, test.staleAxis)
				}
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			workload := test.workload
			if workload == "" {
				workload = testWorkloadID
			}
			_, err = admission.bind(commandConsumerTestContext(t, workload, 3))
			var failure *recordsemantic.Failure
			if !errors.As(err, &failure) ||
				failure.Category != test.want ||
				failure.StaleAxis != test.staleAxis {
				t.Fatalf("admission error = %#v, want %s/%s", err, test.want, test.staleAxis)
			}
		})
	}
}

func TestCommandConsumerAdmissionPropagatesCancellation(t *testing.T) {
	runtimeView := commandFlowAcceptanceTestView(
		t,
		[]catalogsnapshot.CommandExecutionBinding{{
			AuthorizedWorkloadID: testWorkloadID,
			DefinitionRef:        "flow.integration-ingress",
			DefinitionVersion:    "flow-version-1",
			ContextNamespace:     "connector.test",
		}},
	)
	current := commandConsumerIntegrationView(runtimeView)
	admission, err := NewCommandConsumerAdmission(
		t.Context(),
		CommandConsumerSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID: testWorkloadID, RuntimeMetadataView: runtimeView,
		},
		&catalogConsumerViewTestEngine{integrationView: current},
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(
		commandConsumerTestContext(t, testWorkloadID, 3),
	)
	cancel()
	_, err = admission.bind(ctx)
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("canceled admission error = %#v", err)
	}
}

func commandConsumerIntegrationView(
	runtimeView *catalogsnapshot.RuntimeMetadataView,
) catalogsemantic.IntegrationBindingView {
	lineage := runtimeView.Lineage()
	view := catalogsemantic.IntegrationBindingView{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		AuthorizedWorkloadID: testWorkloadID,
		Lineage: catalogsemantic.ViewLineage{
			CatalogRevision:      uint64(lineage.CatalogRevision),
			SchemaGeneration:     uint64(lineage.SchemaGeneration),
			EngineGeneration:     uint64(lineage.EngineGeneration),
			SourceSnapshotDigest: string(runtimeView.SourceSnapshotDigest()),
		},
		Bindings: []catalogsemantic.IntegrationBinding{{
			Kind:               catalogsemantic.BindingConnectorInboundRoute,
			OwnerPack:          "connector.test",
			PackInstallationID: "pack-installation-1",
			PackPayloadDigest:  "nexus-pack-sha256-v1:" + strings.Repeat("a", 64),
			BusinessKey:        "connector.test/inbound",
			DefinitionVersion:  1,
			CredentialRefs:     []string{},
			RuntimeCommand: &catalogsemantic.RuntimeCommandBinding{
				CommandKind:          flowcommandsemantic.CommandKind,
				PayloadFormatVersion: flowcommandsemantic.PayloadFormatVersion,
				DefinitionRef:        "flow.integration-ingress",
				DefinitionVersion:    "flow-version-1",
			},
		}},
		Receipts: catalogsemantic.Receipts{
			ProtocolVersion: "1.0",
			CapabilityVersion: catalogsemantic.
				IntegrationBindingCapabilityName,
			CatalogRevision:  uint64(lineage.CatalogRevision),
			SchemaGeneration: uint64(lineage.SchemaGeneration),
			EngineGeneration: uint64(lineage.EngineGeneration),
			SecurityEpoch:    3,
		},
	}
	view.Lineage.ViewDigest =
		catalogsemantic.ComputeIntegrationBindingViewDigest(view)
	return view
}

func commandConsumerTestContext(
	t *testing.T,
	workloadID string,
	securityEpoch uint64,
) context.Context {
	t.Helper()
	return context.WithValue(
		t.Context(),
		trustedWorkloadCallContextKey{},
		TrustedWorkloadCall{
			orgID: testOrgID, callerWorkloadID: workloadID,
			capability:    commandsemantic.CapabilityName,
			securityEpoch: securityEpoch,
		},
	)
}