package connector_outboundintent

import (
	"context"
	"database/sql/driver"
	"errors"
	"fmt"
	"strings"
	"testing"
	"time"

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

type saturatedIntentSource struct {
	revision   uint64
	lists      int
	acks       int
	generation uint64
}

func (s *saturatedIntentSource) ReadOutboundIntentGeneration(
	_ context.Context,
	request outboundintentv1.GenerationRequest,
) (outboundintentv1.GenerationResponse, error) {
	if s.generation < request.AfterGeneration {
		s.generation = request.AfterGeneration
	}
	if s.generation == 0 {
		s.generation = 1
	}
	return outboundintentv1.GenerationResponse{
		Generation: s.generation, Receipts: validOutboundReceipts(s.revision),
	}, nil
}

func (s *saturatedIntentSource) ListPendingOutboundIntents(
	context.Context,
	outboundintentv1.ListRequest,
) (outboundintentv1.ListResponse, error) {
	s.lists++
	intents := make([]outboundintentv1.Intent, 0, outboundintentv1.MaxIntentsPerList)
	for i := 0; i < outboundintentv1.MaxIntentsPerList; i++ {
		intents = append(intents, outboundIntentTestIntent(
			s.revision, fmt.Sprintf("intent-%d-%d", s.lists, i), fmt.Sprintf("record-%d", i),
		))
	}
	return outboundintentv1.ListResponse{
		Intents: intents, Receipts: validOutboundReceipts(s.revision),
	}, nil
}

func (s *saturatedIntentSource) AcknowledgeOutboundIntent(
	_ context.Context,
	request outboundintentv1.AcknowledgeRequest,
) (outboundintentv1.AcknowledgeResponse, error) {
	s.acks++
	return outboundintentv1.AcknowledgeResponse{
		IntentID: request.IntentID, Outcome: request.Outcome,
		Receipts: validOutboundReceipts(request.CatalogRevision),
	}, nil
}

type acceptAllStore struct {
	accepted int
}

// legacyCommandStore locks the pre-scheduling Worker store shape. Embedding
// the established interface intentionally hides optional methods implemented
// by the concrete test store.
type legacyCommandStore struct {
	commandStore
}

type failingScheduleStore struct {
	*memoryCommandStore
	err error
}

type dependencyReconcileMemoryStore struct {
	*memoryCommandStore
	cancel          context.CancelFunc
	reconcileCalls  int
	reconcileErrors []error
}

func (s *dependencyReconcileMemoryStore) reconcileDependencyWaits(
	context.Context,
	Scope,
	string,
	int,
) (string, bool, int, int, error) {
	s.reconcileCalls++
	if len(s.reconcileErrors) != 0 {
		err := s.reconcileErrors[0]
		s.reconcileErrors = s.reconcileErrors[1:]
		return "", false, 0, 0, err
	}
	if s.command.State == StateSucceeded {
		s.command = testWorkerCommand()
		return "", false, 1, 1, nil
	}
	return "", false, 0, 0, nil
}

func (s *dependencyReconcileMemoryStore) FinalizeWriteback(ctx context.Context, claim Claim) (Command, error) {
	command, err := s.memoryCommandStore.FinalizeWriteback(ctx, claim)
	if err == nil && s.cancel != nil {
		s.cancel()
	}
	return command, err
}

func (s *failingScheduleStore) NextWorkAt(context.Context, Scope) (time.Time, bool, error) {
	return time.Time{}, false, s.err
}

func (*acceptAllStore) VerifySchema(context.Context) error { return nil }

func (s *acceptAllStore) Accept(
	_ context.Context,
	scope Scope,
	intent outboundintentv1.Intent,
) (Command, bool, error) {
	s.accepted++
	return Command{
		CommandID: fmt.Sprintf("command-%d", s.accepted), Scope: scope,
		Intent: intent, State: StatePending,
	}, false, nil
}

func TestServiceFailsClosedBeforeStartupOnSchemaDrift(t *testing.T) {
	scope := outboundIntentTestScope()
	schemaErr := errors.New("schema drift")
	acceptStore := &memoryAcceptStore{schemaErr: schemaErr}
	acceptor, err := NewAcceptor(
		&scriptedIntentSource{intent: outboundIntentTestIntent(scope.CatalogRevision, "intent-1", "record-1")},
		acceptStore,
		scope,
	)
	if err != nil {
		t.Fatal(err)
	}
	workerStore := &memoryCommandStore{command: Command{State: StateSucceeded}}
	worker, err := NewWorker(
		workerStore, &countingPreparer{}, &scriptedCommitter{}, scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	service, err := NewService(acceptor, worker)
	if err != nil {
		t.Fatal(err)
	}
	started := false
	err = service.Run(t.Context(), func() { started = true })
	if !errors.Is(err, schemaErr) || started || acceptStore.verifyCalls != 1 {
		t.Fatalf("schema readiness = err %v started=%v verifies=%d", err, started, acceptStore.verifyCalls)
	}
}

func TestAcceptorSchemaReadinessUsesItsDurableStore(t *testing.T) {
	store := &memoryAcceptStore{}
	acceptor, err := NewAcceptor(
		&scriptedIntentSource{}, store, outboundIntentTestScope(),
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := acceptor.VerifyReady(context.Background()); err != nil || store.verifyCalls != 1 {
		t.Fatalf("VerifyReady = %v calls=%d", err, store.verifyCalls)
	}
}

func TestServicePreservesLegacyPollingWithoutGeneration(t *testing.T) {
	scope := outboundIntentTestScope()
	acceptor, err := NewAcceptor(&legacyIntentSource{scripted: scriptedIntentSource{
		intent:   outboundIntentTestIntent(scope.CatalogRevision, "already-accepted", "record-1"),
		accepted: true,
	}}, &memoryAcceptStore{}, scope)
	if err != nil {
		t.Fatal(err)
	}
	worker, err := NewWorker(
		&memoryCommandStore{}, &countingPreparer{}, &scriptedCommitter{}, scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	service, err := NewService(acceptor, worker)
	if err != nil {
		t.Fatalf("NewService with legacy source = %v, want compatibility", err)
	}
	if service.supportsGenerationScheduling() {
		t.Fatal("legacy source unexpectedly selected generation scheduling")
	}
	ctx, cancel := context.WithCancel(context.Background())
	started := false
	err = service.Run(ctx, func() {
		started = true
		cancel()
	})
	if !started || !errors.Is(err, context.Canceled) {
		t.Fatalf("legacy service = started %v err %v", started, err)
	}
}

func TestServicePreservesLegacyPollingWithoutExactCommandSchedule(t *testing.T) {
	scope := outboundIntentTestScope()
	acceptor, err := NewAcceptor(&scriptedIntentSource{
		intent:   outboundIntentTestIntent(scope.CatalogRevision, "already-accepted", "record-1"),
		accepted: true,
	}, &memoryAcceptStore{}, scope)
	if err != nil {
		t.Fatal(err)
	}
	legacyStore := &legacyCommandStore{commandStore: &memoryCommandStore{}}
	worker, err := NewWorker(
		legacyStore, &countingPreparer{}, &scriptedCommitter{}, scope, "worker-1",
	)
	if err != nil {
		t.Fatalf("NewWorker with legacy store = %v, want compatibility", err)
	}
	service, err := NewService(acceptor, worker)
	if err != nil {
		t.Fatalf("NewService with unscheduled store = %v, want compatibility", err)
	}
	if service.supportsGenerationScheduling() {
		t.Fatal("legacy store unexpectedly selected generation scheduling")
	}
	ctx, cancel := context.WithCancel(context.Background())
	started := false
	err = service.Run(ctx, func() {
		started = true
		cancel()
	})
	if !started || !errors.Is(err, context.Canceled) {
		t.Fatalf("legacy service = started %v err %v", started, err)
	}
}

func TestServiceScheduleReadFailureFallsBackToBoundedPoll(t *testing.T) {
	scope := outboundIntentTestScope()
	worker, err := NewWorker(
		&failingScheduleStore{memoryCommandStore: &memoryCommandStore{}, err: errors.New("schedule unavailable")},
		&countingPreparer{}, &scriptedCommitter{}, scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	service := &Service{worker: worker, legacyPollInterval: 250 * time.Millisecond}
	before := time.Now().Add(service.legacyPollInterval)
	next, scheduled, err := service.drainWorker(t.Context())
	if err != nil || !scheduled || next.Before(before) || next.After(time.Now().Add(time.Second)) {
		t.Fatalf("schedule fallback = next %v scheduled %v err %v", next, scheduled, err)
	}
}

func TestServiceReadinessDoesNotDrainBacklogOrCallProvider(t *testing.T) {
	scope := outboundIntentTestScope()
	source := &saturatedIntentSource{revision: scope.CatalogRevision}
	acceptStore := &acceptAllStore{}
	acceptor, err := NewAcceptor(source, acceptStore, scope)
	if err != nil {
		t.Fatal(err)
	}
	commandStore := &memoryCommandStore{command: testWorkerCommand()}
	worker, err := NewWorker(
		commandStore,
		failingPreparer{err: &PreparationError{
			Code: "association_link_pending", Retryable: true, Cause: ErrDependency,
		}},
		&scriptedCommitter{}, scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	service, err := NewService(acceptor, worker)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	started := false
	runErr := service.Run(ctx, func() {
		started = true
		cancel()
	})
	if !started || !errors.Is(runErr, context.Canceled) {
		t.Fatalf("service startup = started %v err %v", started, runErr)
	}
	if source.lists != 0 || source.acks != 0 || acceptStore.accepted != 0 {
		t.Fatalf("startup acceptance lists=%d acks=%d accepted=%d, want no business work",
			source.lists, source.acks, acceptStore.accepted)
	}
	if commandStore.command.State != StatePending {
		t.Fatalf("provider worker state = %q, want untouched pending work", commandStore.command.State)
	}
}

func TestServiceHandlesPreparationOutcomeAfterReadiness(t *testing.T) {
	scope := outboundIntentTestScope()
	source := &scriptedIntentSource{
		intent: outboundIntentTestIntent(scope.CatalogRevision, "unused", "unused"), accepted: true,
	}
	acceptor, err := NewAcceptor(source, &memoryAcceptStore{}, scope)
	if err != nil {
		t.Fatal(err)
	}
	command := testWorkerCommand()
	store := &memoryCommandStore{command: command}
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	started := false
	worker, err := NewWorker(
		&notifyCommandStore{commandStore: store, changed: func() {
			if !started {
				t.Error("business work preceded readiness")
			}
			cancel()
		}},
		failingPreparer{err: &PreparationError{
			Code: "association_link_pending", Retryable: true, Cause: ErrDependency,
		}},
		&scriptedCommitter{}, scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	service, err := NewService(acceptor, worker)
	if err != nil {
		t.Fatal(err)
	}
	err = service.Run(ctx, func() {
		started = true
	})
	if !started || store.command.State != StateRetry ||
		(err != nil && strings.Contains(err.Error(), "worker startup probe")) {
		t.Fatalf("service = started %v state %q err %v", started, store.command.State, err)
	}
}

func TestServiceIsolatesProviderSuccessIdentityConflictAfterReadiness(t *testing.T) {
	scope := outboundIntentTestScope()
	acceptor, err := NewAcceptor(
		&scriptedIntentSource{
			intent:   outboundIntentTestIntent(scope.CatalogRevision, "already-accepted", "record-1"),
			accepted: true,
		},
		&memoryAcceptStore{},
		scope,
	)
	if err != nil {
		t.Fatal(err)
	}
	command := testWorkerCommand()
	store := &providerSuccessIdentityConflictStore{
		memoryCommandStore: &memoryCommandStore{command: command},
	}
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	started := false
	worker, err := NewWorker(&notifyCommandStore{commandStore: store, changed: func() {
		if !started {
			t.Error("business work preceded readiness")
		}
		cancel()
	}}, &countingPreparer{result: EffectResult{
		Outcome: EffectSucceeded, ResponseStatus: 200, RemoteID: "remote-conflicted",
		LinkFinalization: &RemoteLinkFinalization{RemoteID: "remote-conflicted"},
	}}, &scriptedCommitter{}, command.Scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	service, err := NewService(acceptor, worker)
	if err != nil {
		t.Fatal(err)
	}
	runErr := service.Run(ctx, func() {
		started = true
	})
	if !started || !errors.Is(runErr, context.Canceled) ||
		store.command.State != StateUncertain || store.command.ClaimToken != "" {
		t.Fatalf(
			"service stopped on one conflicting command: started=%v command=%#v err=%v",
			started, store.command, runErr,
		)
	}
}

func TestServiceUnchangedGenerationDoesNotRescanRuntimeQueue(t *testing.T) {
	scope := outboundIntentTestScope()
	source := &scriptedIntentSource{
		intent:   outboundIntentTestIntent(scope.CatalogRevision, "already-accepted", "record-1"),
		accepted: true, generation: 1, generationRead: make(chan struct{}, 16),
	}
	acceptor, err := NewAcceptor(source, &memoryAcceptStore{}, scope)
	if err != nil {
		t.Fatal(err)
	}
	worker, err := NewWorker(
		&memoryCommandStore{command: Command{State: StateSucceeded}},
		&countingPreparer{}, &scriptedCommitter{}, scope, "worker-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	service, err := NewService(acceptor, worker)
	if err != nil {
		t.Fatal(err)
	}
	service.generationProbeInterval = 2 * time.Millisecond
	service.recoveryReconcileInterval = time.Hour

	ctx, cancel := context.WithCancel(context.Background())
	started := make(chan struct{})
	done := make(chan error, 1)
	go func() { done <- service.Run(ctx, func() { close(started) }) }()
	<-started
	for range 3 {
		select {
		case <-source.generationRead:
		case <-time.After(time.Second):
			t.Fatal("generation probe did not run")
		}
	}
	cancel()
	if runErr := <-done; !errors.Is(runErr, context.Canceled) {
		t.Fatalf("Run() error = %v, want cancellation", runErr)
	}
	if source.generationReads < 3 || source.listCalls != 1 {
		t.Fatalf("generation reads=%d queue lists=%d, want >=3 reads and only startup list", source.generationReads, source.listCalls)
	}
}

func TestServiceDependencySafetyTickWakesAndDrainsWorker(t *testing.T) {
	scope := outboundIntentTestScope()
	acceptor, err := NewAcceptor(&scriptedIntentSource{
		intent:   outboundIntentTestIntent(scope.CatalogRevision, "already-accepted", "record-1"),
		accepted: true, generation: 1,
	}, &memoryAcceptStore{}, scope)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	store := &dependencyReconcileMemoryStore{
		memoryCommandStore: &memoryCommandStore{command: Command{State: StateSucceeded}},
		cancel:             cancel,
	}
	worker, err := NewWorker(store, &countingPreparer{result: EffectResult{
		Outcome: EffectSucceeded, ResponseStatus: 200, RemoteID: "remote-safety-wake",
		LinkFinalization: &RemoteLinkFinalization{RemoteID: "remote-safety-wake"},
	}}, &scriptedCommitter{}, scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	service, err := NewService(acceptor, worker)
	if err != nil {
		t.Fatal(err)
	}
	service.generationProbeInterval = time.Hour
	service.recoveryReconcileInterval = time.Hour
	service.dependencyReconcileInterval = 2 * time.Millisecond

	started := false
	runErr := service.Run(ctx, func() { started = true })
	if !started || !errors.Is(runErr, context.Canceled) || store.reconcileCalls != 1 ||
		store.command.State != StateSucceeded {
		t.Fatalf("dependency safety tick = started=%v calls=%d state=%q err=%v",
			started, store.reconcileCalls, store.command.State, runErr)
	}
}

func TestLegacyServiceDependencySafetyTickWakesAndDrainsWorker(t *testing.T) {
	scope := outboundIntentTestScope()
	acceptor, err := NewAcceptor(&legacyIntentSource{scripted: scriptedIntentSource{
		intent:   outboundIntentTestIntent(scope.CatalogRevision, "already-accepted", "record-1"),
		accepted: true,
	}}, &memoryAcceptStore{}, scope)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	store := &dependencyReconcileMemoryStore{
		memoryCommandStore: &memoryCommandStore{command: Command{State: StateSucceeded}},
		cancel:             cancel,
	}
	worker, err := NewWorker(store, &countingPreparer{result: EffectResult{
		Outcome: EffectSucceeded, ResponseStatus: 200, RemoteID: "remote-legacy-safety-wake",
		LinkFinalization: &RemoteLinkFinalization{RemoteID: "remote-legacy-safety-wake"},
	}}, &scriptedCommitter{}, scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	service, err := NewService(acceptor, worker)
	if err != nil {
		t.Fatal(err)
	}
	service.legacyPollInterval = 2 * time.Millisecond
	service.dependencyReconcileInterval = 2 * time.Millisecond

	started := false
	runErr := service.Run(ctx, func() { started = true })
	if !started || !errors.Is(runErr, context.Canceled) || store.reconcileCalls != 1 ||
		store.command.State != StateSucceeded {
		t.Fatalf("legacy dependency safety tick = started=%v calls=%d state=%q err=%v",
			started, store.reconcileCalls, store.command.State, runErr)
	}
}

func TestServiceDependencySafetyTickSurvivesTransientDatabaseFailure(t *testing.T) {
	scope := outboundIntentTestScope()
	acceptor, err := NewAcceptor(&scriptedIntentSource{
		intent:   outboundIntentTestIntent(scope.CatalogRevision, "already-accepted", "record-1"),
		accepted: true, generation: 1,
	}, &memoryAcceptStore{}, scope)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
	defer cancel()
	store := &dependencyReconcileMemoryStore{
		memoryCommandStore: &memoryCommandStore{command: Command{State: StateSucceeded}},
		cancel:             cancel, reconcileErrors: []error{driver.ErrBadConn},
	}
	worker, err := NewWorker(store, &countingPreparer{result: EffectResult{
		Outcome: EffectSucceeded, ResponseStatus: 200, RemoteID: "remote-after-transient-reconcile",
		LinkFinalization: &RemoteLinkFinalization{RemoteID: "remote-after-transient-reconcile"},
	}}, &scriptedCommitter{}, scope, "worker-1")
	if err != nil {
		t.Fatal(err)
	}
	service, err := NewService(acceptor, worker)
	if err != nil {
		t.Fatal(err)
	}
	service.generationProbeInterval = time.Hour
	service.recoveryReconcileInterval = time.Hour
	service.dependencyReconcileInterval = 2 * time.Millisecond

	started := false
	runErr := service.Run(ctx, func() { started = true })
	if !started || !errors.Is(runErr, context.Canceled) || store.reconcileCalls != 2 ||
		store.command.State != StateSucceeded {
		t.Fatalf("transient dependency safety tick = started=%v calls=%d state=%q err=%v",
			started, store.reconcileCalls, store.command.State, runErr)
	}
}

// Simulate an indeterminate database response: readiness may have committed,
// but the safety call blocks until its budget expires and reports no wake.
type stalledDependencyStore struct {
	*dependencyReconcileMemoryStore
}

func (s *stalledDependencyStore) reconcileDependencyWaits(ctx context.Context, _ Scope, _ string, _ int) (string, bool, int, int, error) {
	if s.reconcileCalls > 0 {
		return "", false, 0, 0, nil
	}
	s.reconcileCalls++
	<-ctx.Done()
	s.command = testWorkerCommand()
	return "last-completed-waiter", true, 0, 1, ctx.Err()
}

func TestDependencySafetyBudgetYieldsToNormalWorkAcrossEntrypoints(t *testing.T) {
	for _, entry := range []string{"service", "legacy", "worker"} {
		t.Run(entry, func(t *testing.T) {
			ctx, cancel := context.WithTimeout(t.Context(), time.Second)
			defer cancel()
			scope := outboundIntentTestScope()
			store := &stalledDependencyStore{&dependencyReconcileMemoryStore{
				memoryCommandStore: &memoryCommandStore{command: Command{State: StateSucceeded}}, cancel: cancel,
			}}
			worker, err := NewWorker(store, &countingPreparer{result: EffectResult{
				Outcome: EffectSucceeded, ResponseStatus: 200, RemoteID: "remote-after-budget",
				LinkFinalization: &RemoteLinkFinalization{RemoteID: "remote-after-budget"},
			}}, &scriptedCommitter{}, scope, "budget-worker")
			if err != nil {
				t.Fatal(err)
			}
			worker.dependencyReconcileBudget = 20 * time.Millisecond
			worker.dependencyReconcileInterval = 2 * time.Millisecond
			var runErr error
			if entry == "worker" {
				runErr = worker.Run(ctx, func() {})
			} else {
				source := &scriptedIntentSource{intent: outboundIntentTestIntent(scope.CatalogRevision, "accepted", "record"), accepted: true, generation: 1}
				acceptor, err := NewAcceptor(source, &memoryAcceptStore{}, scope)
				if err != nil {
					t.Fatal(err)
				}
				if entry == "legacy" {
					acceptor, err = NewAcceptor(&legacyIntentSource{scripted: *source}, &memoryAcceptStore{}, scope)
					if err != nil {
						t.Fatal(err)
					}
				}
				service, err := NewService(acceptor, worker)
				if err != nil {
					t.Fatal(err)
				}
				service.generationProbeInterval = time.Hour
				service.recoveryReconcileInterval = time.Hour
				service.dependencyReconcileInterval = 2 * time.Millisecond
				service.legacyPollInterval = 2 * time.Millisecond
				runErr = service.Run(ctx, func() {})
			}
			if !errors.Is(runErr, context.Canceled) || store.command.State != StateSucceeded {
				t.Fatalf("budget stopped or starved owner: state=%s err=%v", store.command.State, runErr)
			}
		})
	}
}