package connector_outboundintent

import (
	"context"
	"errors"
	"testing"

	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

type scriptedIntentSource struct {
	intent          outboundintentv1.Intent
	generation      uint64
	generationReads int
	generationRead  chan struct{}
	listCalls       int
	ackCalls        int
	failFirstAck    bool
	accepted        bool
}

// legacyIntentSource locks the pre-generation exported interface shape. It
// deliberately has no ReadOutboundIntentGeneration method.
type legacyIntentSource struct {
	scripted scriptedIntentSource
}

var _ IntentSource = (*legacyIntentSource)(nil)

func (s *legacyIntentSource) ListPendingOutboundIntents(
	ctx context.Context,
	request outboundintentv1.ListRequest,
) (outboundintentv1.ListResponse, error) {
	return s.scripted.ListPendingOutboundIntents(ctx, request)
}

func (s *legacyIntentSource) AcknowledgeOutboundIntent(
	ctx context.Context,
	request outboundintentv1.AcknowledgeRequest,
) (outboundintentv1.AcknowledgeResponse, error) {
	return s.scripted.AcknowledgeOutboundIntent(ctx, request)
}

func (s *scriptedIntentSource) ReadOutboundIntentGeneration(
	_ context.Context,
	request outboundintentv1.GenerationRequest,
) (outboundintentv1.GenerationResponse, error) {
	s.generationReads++
	if s.generationRead != nil {
		select {
		case s.generationRead <- struct{}{}:
		default:
		}
	}
	if s.generation == 0 && s.intent.IntentID != "" {
		s.generation = 1
	}
	generation := s.generation
	if generation < request.AfterGeneration {
		generation = request.AfterGeneration
	}
	return outboundintentv1.GenerationResponse{
		Generation: generation,
		Receipts:   validOutboundReceipts(s.intent.CatalogRevision),
	}, nil
}

func (s *scriptedIntentSource) ListPendingOutboundIntents(_ context.Context, _ outboundintentv1.ListRequest) (outboundintentv1.ListResponse, error) {
	s.listCalls++
	intents := []outboundintentv1.Intent(nil)
	if !s.accepted {
		intents = []outboundintentv1.Intent{s.intent}
	}
	return outboundintentv1.ListResponse{Intents: intents, Receipts: validOutboundReceipts(s.intent.CatalogRevision)}, nil
}

func (s *scriptedIntentSource) AcknowledgeOutboundIntent(_ context.Context, request outboundintentv1.AcknowledgeRequest) (outboundintentv1.AcknowledgeResponse, error) {
	s.ackCalls++
	if s.failFirstAck && s.ackCalls == 1 {
		return outboundintentv1.AcknowledgeResponse{}, errors.New("lost acknowledgement response")
	}
	s.accepted = true
	return outboundintentv1.AcknowledgeResponse{
		IntentID: request.IntentID, Outcome: request.Outcome,
		Receipts: validOutboundReceipts(request.CatalogRevision),
	}, nil
}

type memoryAcceptStore struct {
	command     Command
	calls       int
	verifyCalls int
	schemaErr   error
}

func (s *memoryAcceptStore) VerifySchema(context.Context) error {
	s.verifyCalls++
	return s.schemaErr
}

func (s *memoryAcceptStore) Accept(_ context.Context, scope Scope, intent outboundintentv1.Intent) (Command, bool, error) {
	s.calls++
	if s.command.CommandID == "" {
		s.command = Command{CommandID: "command-1", Scope: scope, Intent: intent, State: StatePending}
		return s.command, false, nil
	}
	if s.command.Scope != scope || s.command.Intent.IntentID != intent.IntentID {
		return Command{}, true, ErrIdentityConflict
	}
	return s.command, true, nil
}

func TestAcceptorPersistsBeforeAcknowledgementAndRecoversLostResponse(t *testing.T) {
	scope := Scope{OrgID: "org-1", AppInstallationID: "app-1", AuthorizedWorkloadID: "integration-1", CatalogRevision: 41}
	intent := outboundIntentTestIntent(scope.CatalogRevision, "intent-1", "record-1")
	source := &scriptedIntentSource{intent: intent, failFirstAck: true}
	store := &memoryAcceptStore{}
	acceptor, err := NewAcceptor(source, store, scope)
	if err != nil {
		t.Fatal(err)
	}
	if processed, err := acceptor.ProcessBatch(t.Context()); err == nil || processed != 0 {
		t.Fatalf("first batch = (%d, %v), want lost response after durable acceptance", processed, err)
	}
	if store.calls != 1 || store.command.CommandID == "" || source.ackCalls != 1 {
		t.Fatalf("first batch order: accepts=%d command=%q acknowledgements=%d", store.calls, store.command.CommandID, source.ackCalls)
	}
	if processed, err := acceptor.ProcessBatch(t.Context()); err != nil || processed != 1 {
		t.Fatalf("retry batch = (%d, %v)", processed, err)
	}
	if store.calls != 2 || source.ackCalls != 2 || !source.accepted {
		t.Fatalf("retry did not close handoff: accepts=%d acknowledgements=%d accepted=%v", store.calls, source.ackCalls, source.accepted)
	}
}

func TestAcceptorKeepsLegacyIntentSourceCompatibility(t *testing.T) {
	scope := outboundIntentTestScope()
	source := &legacyIntentSource{scripted: scriptedIntentSource{
		intent: outboundIntentTestIntent(scope.CatalogRevision, "intent-legacy", "record-1"),
	}}
	acceptor, err := NewAcceptor(source, &memoryAcceptStore{}, scope)
	if err != nil {
		t.Fatal(err)
	}
	if processed, err := acceptor.ProcessBatch(t.Context()); err != nil || processed != 1 {
		t.Fatalf("legacy source batch = (%d, %v), want (1, nil)", processed, err)
	}
	if _, err := acceptor.ReadGeneration(t.Context(), 0); !errors.Is(err, ErrDependency) {
		t.Fatalf("legacy source generation read = %v, want dependency error", err)
	}
}

func validOutboundReceipts(revision uint64) outboundintentv1.Receipts {
	return outboundintentv1.Receipts{
		ProtocolVersion: "1.0", CapabilityVersion: outboundintentv1.CapabilityName,
		CatalogRevision: revision, SchemaGeneration: 7, EngineGeneration: 9, SecurityEpoch: 11,
	}
}

type invalidReceiptIntentSource struct{ scriptedIntentSource }

func (s *invalidReceiptIntentSource) ListPendingOutboundIntents(_ context.Context, _ outboundintentv1.ListRequest) (outboundintentv1.ListResponse, error) {
	return outboundintentv1.ListResponse{
		Intents:  []outboundintentv1.Intent{s.intent},
		Receipts: outboundintentv1.Receipts{CatalogRevision: s.intent.CatalogRevision},
	}, nil
}

func TestAcceptorRejectsIncompleteRuntimeReceiptsBeforeDurableAcceptance(t *testing.T) {
	scope := outboundIntentTestScope()
	source := &invalidReceiptIntentSource{scriptedIntentSource: scriptedIntentSource{
		intent: outboundIntentTestIntent(scope.CatalogRevision, "intent-invalid-receipts", "record-1"),
	}}
	store := &memoryAcceptStore{}
	acceptor, err := NewAcceptor(source, store, scope)
	if err != nil {
		t.Fatal(err)
	}
	if processed, err := acceptor.ProcessBatch(t.Context()); !errors.Is(err, ErrIdentityConflict) || processed != 0 {
		t.Fatalf("invalid receipts batch = (%d, %v)", processed, err)
	}
	if store.calls != 0 || source.ackCalls != 0 {
		t.Fatalf("invalid receipts crossed durable handoff: accepts=%d acknowledgements=%d", store.calls, source.ackCalls)
	}
}