package runtimehandoff

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

	connector_inbound "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	commandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	commonv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	flowcommandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv1"
	flowcommandv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
)

type platformClientStub struct {
	view        catalogv1.IntegrationBindingView
	viewErr     error
	acceptErr   error
	response    commandv1.Response
	viewCalls   int
	acceptCalls []commandv1.Request
	callOrder   *[]string
}

func (c *platformClientStub) GetIntegrationBindingView(
	_ context.Context,
	_ catalogv1.GetRequest,
) (catalogv1.IntegrationBindingView, error) {
	c.viewCalls++
	return c.view, c.viewErr
}

func (c *platformClientStub) AcceptCommand(
	_ context.Context,
	request commandv1.Request,
) (commandv1.Response, error) {
	if c.callOrder != nil {
		*c.callOrder = append(*c.callOrder, "accept")
	}
	c.acceptCalls = append(c.acceptCalls, request)
	if c.acceptErr != nil {
		return commandv1.Response{}, c.acceptErr
	}
	response := c.response
	if response.CommandID == "" {
		response.CommandID = request.CommandID
	}
	if response.Receipts.CapabilityVersion == "" {
		response.Receipts = validCommandReceipts(
			*request.MinCatalogRevision,
		)
	}
	return response, nil
}

type memoryReceiptStore struct {
	prepared     *PreparedCommand
	loadCalls    int
	prepareCalls int
	callOrder    *[]string
	verifyErr    error
	loadErr      error
}

func (s *memoryReceiptStore) VerifySchema(context.Context) error {
	return s.verifyErr
}

func (s *memoryReceiptStore) LoadForScope(
	_ context.Context,
	_ connector_inbound.RuntimeCommandHandoffScope,
) (PreparedCommand, bool, error) {
	s.loadCalls++
	if s.loadErr != nil {
		return PreparedCommand{}, false, s.loadErr
	}
	if s.prepared == nil {
		return PreparedCommand{}, false, nil
	}
	return *s.prepared, true, nil
}

func (s *memoryReceiptStore) Prepare(
	_ context.Context,
	_ connector_inbound.RuntimeCommandHandoffScope,
	candidate PreparedCommand,
) (PreparedCommand, bool, error) {
	s.prepareCalls++
	if s.callOrder != nil {
		*s.callOrder = append(*s.callOrder, "prepare")
	}
	if s.prepared != nil {
		return *s.prepared, true, nil
	}
	copy := candidate
	s.prepared = &copy
	return copy, false, nil
}

func TestHandoffPersistsExactCommandBeforeRuntimeAcceptance(t *testing.T) {
	order := []string{}
	client := &platformClientStub{
		view: validBindingView(), callOrder: &order,
	}
	store := &memoryReceiptStore{callOrder: &order}
	handoff := newTestHandoff(t, client, store)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	request := testHandoffRequest()
	request.DependencySource = flowcommandv2.DependencySourceScope{
		ObjectKind: "contacts", SourceID: "remote-contact-1",
	}

	if err := handoff.AcceptRuntimeCommand(ctx, request); err != nil {
		t.Fatalf("AcceptRuntimeCommand() error = %v", err)
	}
	if strings.Join(order, ",") != "prepare,accept" ||
		store.prepareCalls != 1 || len(client.acceptCalls) != 1 {
		t.Fatalf(
			"order=%v prepare=%d accepts=%d",
			order,
			store.prepareCalls,
			len(client.acceptCalls),
		)
	}
	prepared := *store.prepared
	if prepared.MinimumCatalogRevision != 41 ||
		prepared.BindingViewDigest != client.view.Lineage.ViewDigest ||
		prepared.Request.CommandID == "" ||
		prepared.Request.CommandID != prepared.Request.IdempotencyKey ||
		prepared.Request.PartitionKey == request.Scope.SourcePartition {
		t.Fatalf("prepared command = %#v", prepared)
	}
	payload, err := flowcommandv1.Decode(prepared.Request.Payload)
	if err != nil {
		t.Fatal(err)
	}
	if payload.DefinitionRef != request.DefinitionRef ||
		payload.DefinitionVersion != request.DefinitionVersion ||
		string(payload.CanonicalInputs) != `{"eventID":"event-1","record":{"Name":"Ada"}}` {
		t.Fatalf("payload = %#v", payload)
	}
}

func TestHandoffUsesV2OnlyWhenCatalogAuthorizesDependencySource(t *testing.T) {
	view := validBindingView()
	view.Bindings[0].RuntimeCommand.PayloadFormatVersion = flowcommandv2.PayloadFormatVersion
	view.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(view)
	client := &platformClientStub{view: view}
	store := &memoryReceiptStore{}
	handoff := newTestHandoff(t, client, store)
	request := testHandoffRequest()
	request.DependencySource = flowcommandv2.DependencySourceScope{
		ObjectKind: "contacts",
		SourceID:   "remote-contact-1",
	}

	if err := handoff.AcceptRuntimeCommand(t.Context(), request); err != nil {
		t.Fatalf("AcceptRuntimeCommand() error = %v", err)
	}
	if store.prepared == nil || store.prepared.Request.PayloadFormatVersion != flowcommandv2.PayloadFormatVersion ||
		len(client.acceptCalls) != 1 {
		t.Fatalf("prepared=%#v accepts=%d", store.prepared, len(client.acceptCalls))
	}
	payload, err := flowcommandv2.Decode(store.prepared.Request.Payload)
	if err != nil {
		t.Fatal(err)
	}
	if payload.DefinitionRef != request.DefinitionRef ||
		payload.DefinitionVersion != request.DefinitionVersion ||
		payload.DependencySource != request.DependencySource ||
		string(payload.CanonicalInputs) != `{"eventID":"event-1","record":{"Name":"Ada"}}` {
		t.Fatalf("V2 payload = %#v", payload)
	}
}

func TestHandoffRejectsIncompleteV2SourceBeforePreparingReceipt(t *testing.T) {
	view := validBindingView()
	view.Bindings[0].RuntimeCommand.PayloadFormatVersion = flowcommandv2.PayloadFormatVersion
	view.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(view)
	client := &platformClientStub{view: view}
	store := &memoryReceiptStore{}
	handoff := newTestHandoff(t, client, store)

	err := handoff.AcceptRuntimeCommand(t.Context(), testHandoffRequest())
	if err == nil || store.prepareCalls != 0 || len(client.acceptCalls) != 0 {
		t.Fatalf("error=%v prepare=%d accepts=%d", err, store.prepareCalls, len(client.acceptCalls))
	}
	var failure *connector_inbound.RuntimeCommandHandoffFailure
	if !errors.As(err, &failure) || failure.Retriable {
		t.Fatalf("failure = %#v", err)
	}
}

func TestHandoffResumeReusesStoredBytesWithoutCatalogOrRebuild(t *testing.T) {
	prepared := testPreparedCommand([]byte(`{"stored":true}`))
	prepared.OrgID = "org-1"
	prepared.AppInstallationID = "01900000-0000-7000-8000-000000000001"
	prepared.AuthorizedWorkloadID = "integration"
	prepared.ScopeDigest = consumerScopeDigest(
		"org-1",
		"01900000-0000-7000-8000-000000000001",
		"integration",
	)
	prepared.Request.CommandID = stableIdentity("integration-workitem", "work-1")
	prepared.Request.IdempotencyKey = prepared.Request.CommandID
	prepared.Request.PartitionKey = stableIdentity("integration-source", "source-1")
	client := &platformClientStub{}
	store := &memoryReceiptStore{prepared: &prepared}
	handoff := newTestHandoff(t, client, store)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	found, err := handoff.ResumePreparedRuntimeCommand(ctx, testScope())
	if err != nil || !found {
		t.Fatalf("ResumePreparedRuntimeCommand() = %t, %v", found, err)
	}
	if client.viewCalls != 0 || store.prepareCalls != 0 ||
		len(client.acceptCalls) != 1 ||
		string(client.acceptCalls[0].Payload) != `{"stored":true}` {
		t.Fatalf(
			"view calls=%d prepare=%d accepts=%#v",
			client.viewCalls,
			store.prepareCalls,
			client.acceptCalls,
		)
	}
}

func TestHandoffResumeReusesStoredV2BytesWithoutReauthoringSource(t *testing.T) {
	payload, err := flowcommandv2.Encode(flowcommandv2.Payload{
		DefinitionRef:     "Companion_Flow",
		DefinitionVersion: "flow-version-7",
		CanonicalInputs:   []byte(`{"eventID":"event-1"}`),
		DependencySource: flowcommandv2.DependencySourceScope{
			ObjectKind: "contacts", SourceID: "remote-contact-1",
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	prepared := testPreparedCommand(payload)
	prepared.OrgID = "org-1"
	prepared.AppInstallationID = "01900000-0000-7000-8000-000000000001"
	prepared.AuthorizedWorkloadID = "integration"
	prepared.ScopeDigest = consumerScopeDigest(
		prepared.OrgID, prepared.AppInstallationID, prepared.AuthorizedWorkloadID,
	)
	prepared.Request.CommandID = stableIdentity("integration-workitem", "work-1")
	prepared.Request.IdempotencyKey = prepared.Request.CommandID
	prepared.Request.PartitionKey = stableIdentity("integration-source", "source-1")
	prepared.Request.PayloadFormatVersion = flowcommandv2.PayloadFormatVersion
	client := &platformClientStub{}
	handoff := newTestHandoff(t, client, &memoryReceiptStore{prepared: &prepared})

	found, err := handoff.ResumePreparedRuntimeCommand(t.Context(), testScope())
	if err != nil || !found || client.viewCalls != 0 || len(client.acceptCalls) != 1 ||
		client.acceptCalls[0].PayloadDigest != commandv1.ComputePayloadDigest(payload) ||
		string(client.acceptCalls[0].Payload) != string(payload) {
		t.Fatalf("resume found=%t err=%v view=%d accepts=%#v", found, err, client.viewCalls, client.acceptCalls)
	}
}

func TestHandoffFailsClosedBeforeReceiptWithoutExactRouteBinding(t *testing.T) {
	view := validBindingView()
	view.Bindings[0].DefinitionVersion++
	view.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(view)
	client := &platformClientStub{view: view}
	store := &memoryReceiptStore{}
	handoff := newTestHandoff(t, client, store)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	err := handoff.AcceptRuntimeCommand(ctx, testHandoffRequest())
	if err == nil || store.prepareCalls != 0 || len(client.acceptCalls) != 0 {
		t.Fatalf(
			"error=%v prepare=%d accepts=%d",
			err,
			store.prepareCalls,
			len(client.acceptCalls),
		)
	}
	var failure *connector_inbound.RuntimeCommandHandoffFailure
	if !errors.As(err, &failure) || failure.Retriable {
		t.Fatalf("failure = %#v", err)
	}
}

func TestHandoffUsesCatalogExactVersionWhenEventHasNoCompanionVersion(t *testing.T) {
	client := &platformClientStub{view: validBindingView()}
	store := &memoryReceiptStore{}
	handoff := newTestHandoff(t, client, store)
	request := testHandoffRequest()
	request.DefinitionVersion = ""
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	if err := handoff.AcceptRuntimeCommand(ctx, request); err != nil {
		t.Fatalf("AcceptRuntimeCommand() error = %v", err)
	}
	if store.prepared == nil || len(client.acceptCalls) != 1 {
		t.Fatalf("prepared=%#v accepts=%d", store.prepared, len(client.acceptCalls))
	}
	payload, err := flowcommandv1.Decode(store.prepared.Request.Payload)
	if err != nil {
		t.Fatal(err)
	}
	if payload.DefinitionVersion != "flow-version-7" {
		t.Fatalf("definition version = %q", payload.DefinitionVersion)
	}
}

func TestHandoffClassifiesTransientPlatformFailureForWorkItemRetry(t *testing.T) {
	client := &platformClientStub{viewErr: &commonv1.Failure{
		Category: commonv1.FailureUnavailable,
	}}
	handoff := newTestHandoff(t, client, &memoryReceiptStore{})
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	err := handoff.AcceptRuntimeCommand(ctx, testHandoffRequest())
	var failure *connector_inbound.RuntimeCommandHandoffFailure
	if !errors.As(err, &failure) || !failure.Retriable {
		t.Fatalf("failure = %#v", err)
	}
}

func TestHandoffClassifiesIntegrationStoreFailureForWorkItemRetry(t *testing.T) {
	client := &platformClientStub{}
	store := &memoryReceiptStore{loadErr: &storeUnavailableError{
		Cause: errors.New("TiDB unavailable"),
	}}
	handoff := newTestHandoff(t, client, store)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	err := handoff.AcceptRuntimeCommand(ctx, testHandoffRequest())
	var failure *connector_inbound.RuntimeCommandHandoffFailure
	if !errors.As(err, &failure) || !failure.Retriable || client.viewCalls != 0 {
		t.Fatalf("failure = %#v view calls=%d", err, client.viewCalls)
	}
}

func TestHandoffRejectsStoredReceiptOutsideConsumerScopeBeforeAccept(t *testing.T) {
	prepared := testPreparedCommand([]byte(`{"stored":true}`))
	prepared.OrgID = "other-org"
	prepared.AppInstallationID = "01900000-0000-7000-8000-000000000001"
	prepared.AuthorizedWorkloadID = "integration"
	prepared.ScopeDigest = consumerScopeDigest(
		prepared.OrgID,
		prepared.AppInstallationID,
		prepared.AuthorizedWorkloadID,
	)
	prepared.Request.CommandID = stableIdentity("integration-workitem", "work-1")
	prepared.Request.IdempotencyKey = prepared.Request.CommandID
	prepared.Request.PartitionKey = stableIdentity("integration-source", "source-1")
	client := &platformClientStub{}
	handoff := newTestHandoff(t, client, &memoryReceiptStore{prepared: &prepared})
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	found, err := handoff.ResumePreparedRuntimeCommand(ctx, testScope())
	if !found || err == nil || len(client.acceptCalls) != 0 {
		t.Fatalf("resume found=%t error=%v accepts=%d", found, err, len(client.acceptCalls))
	}
}

func TestHandoffRejectsStaleRuntimeAcceptanceReceipt(t *testing.T) {
	client := &platformClientStub{
		view:     validBindingView(),
		response: commandv1.Response{Receipts: validCommandReceipts(40)},
	}
	handoff := newTestHandoff(t, client, &memoryReceiptStore{})
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	err := handoff.AcceptRuntimeCommand(ctx, testHandoffRequest())
	if err == nil || !strings.Contains(err.Error(), "stale Catalog revision") {
		t.Fatalf("AcceptRuntimeCommand() error = %v", err)
	}
	var failure *connector_inbound.RuntimeCommandHandoffFailure
	if !errors.As(err, &failure) || !failure.Retriable {
		t.Fatalf("stale acceptance outcome must remain retryable: %#v", err)
	}
}

func TestHandoffKeepsMalformedAcceptanceResponseNonTerminal(t *testing.T) {
	client := &platformClientStub{
		view: validBindingView(),
		response: commandv1.Response{
			CommandID: "different-command",
			Receipts:  validCommandReceipts(41),
		},
	}
	handoff := newTestHandoff(t, client, &memoryReceiptStore{})
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	err := handoff.AcceptRuntimeCommand(ctx, testHandoffRequest())
	var failure *connector_inbound.RuntimeCommandHandoffFailure
	if !errors.As(err, &failure) || !failure.Retriable ||
		!strings.Contains(err.Error(), "validate Runtime command acceptance") {
		t.Fatalf("malformed acceptance outcome = %#v", err)
	}
}

func TestConsumerScopeDigestSeparatesEveryAuthorityAxis(t *testing.T) {
	base := consumerScopeDigest(
		"org-1",
		"01900000-0000-7000-8000-000000000001",
		"integration",
	)
	if len(base) != 64 {
		t.Fatalf("scope digest length = %d", len(base))
	}
	for name, candidate := range map[string]string{
		"org": consumerScopeDigest(
			"org-2",
			"01900000-0000-7000-8000-000000000001",
			"integration",
		),
		"installation": consumerScopeDigest(
			"org-1",
			"01900000-0000-7000-8000-000000000002",
			"integration",
		),
		"workload": consumerScopeDigest(
			"org-1",
			"01900000-0000-7000-8000-000000000001",
			"integration-2",
		),
	} {
		if candidate == base {
			t.Fatalf("%s axis did not change scope digest", name)
		}
	}
}

func newTestHandoff(
	t *testing.T,
	client PlatformClient,
	store ReceiptStore,
) *Handoff {
	t.Helper()
	handoff, err := NewHandoff(client, store, Config{
		OrgID:                "org-1",
		AppInstallationID:    "01900000-0000-7000-8000-000000000001",
		AuthorizedWorkloadID: "integration",
	})
	if err != nil {
		t.Fatal(err)
	}
	return handoff
}

func testHandoffRequest() connector_inbound.RuntimeCommandHandoffRequest {
	return connector_inbound.RuntimeCommandHandoffRequest{
		Scope:             testScope(),
		DefinitionRef:     "Companion_Flow",
		DefinitionVersion: "flow-version-7",
		Inputs: map[string]any{
			"record":  map[string]any{"Name": "Ada"},
			"eventID": "event-1",
		},
	}
}

func validBindingView() catalogv1.IntegrationBindingView {
	view := catalogv1.IntegrationBindingView{
		OrgID:                "org-1",
		AppInstallationID:    "01900000-0000-7000-8000-000000000001",
		AuthorizedWorkloadID: "integration",
		Lineage: catalogv1.ViewLineage{
			CatalogRevision: 41, SchemaGeneration: 7, EngineGeneration: 8,
			SourceSnapshotDigest: "sha256:" + strings.Repeat("b", 64),
		},
		Bindings: []catalogv1.IntegrationBinding{{
			Kind:      catalogv1.BindingConnectorInboundRoute,
			OwnerPack: "connector-pack", PackInstallationID: "install-1",
			PackPayloadDigest: "nexus-pack-sha256-v1:" + strings.Repeat("a", 64),
			BusinessKey:       "connector-pack/contact.created", DefinitionVersion: 7,
			CredentialRefs: []string{},
			RuntimeCommand: &catalogv1.RuntimeCommandBinding{
				CommandKind:          flowcommandv1.CommandKind,
				PayloadFormatVersion: flowcommandv1.PayloadFormatVersion,
				DefinitionRef:        "Companion_Flow", DefinitionVersion: "flow-version-7",
			},
		}},
		Receipts: catalogv1.Receipts{
			ProtocolVersion:   "1.0",
			CapabilityVersion: catalogv1.IntegrationBindingCapabilityName,
			CatalogRevision:   41, SchemaGeneration: 7, EngineGeneration: 8,
		},
	}
	view.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(view)
	return view
}

func validCommandReceipts(revision uint64) commandv1.Receipts {
	return commandv1.Receipts{
		ProtocolVersion: "1.0", CapabilityVersion: commandv1.CapabilityName,
		CatalogRevision: revision, SchemaGeneration: 7, EngineGeneration: 8,
	}
}

var _ PlatformClient = (*platformClientStub)(nil)
var _ ReceiptStore = (*memoryReceiptStore)(nil)