package connector_outboundintent

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

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

type firstClaimFailureStore struct {
	*memoryCommandStore
	calls int
	err   error
}

func (s *firstClaimFailureStore) ClaimWriteback(ctx context.Context, scope Scope, worker, token string, lease time.Duration) (Command, bool, error) {
	s.calls++
	if s.calls == 1 {
		return Command{}, false, s.err
	}
	return s.memoryCommandStore.ClaimWriteback(ctx, scope, worker, token, lease)
}

func TestStandaloneWorkerRetainsFirstTurnReadinessContract(t *testing.T) {
	for _, err := range []error{driver.ErrBadConn, ErrIdentityConflict} {
		store := &firstClaimFailureStore{memoryCommandStore: &memoryCommandStore{}, err: err}
		worker, _ := NewWorker(store, &countingPreparer{}, &scriptedCommitter{}, outboundIntentTestScope(), "standalone")
		ctx, cancel := context.WithTimeout(t.Context(), time.Second)
		started := false
		result := worker.Run(ctx, func() { started = true; cancel() })
		cancel()
		if err == ErrIdentityConflict {
			if started || store.calls != 1 || !errors.Is(result, err) {
				t.Fatalf("permanent startup: started=%t calls=%d err=%v", started, store.calls, result)
			}
		} else {
			if !started || store.calls != 2 || !errors.Is(result, context.Canceled) {
				t.Fatalf("transient startup: started=%t calls=%d err=%v", started, store.calls, result)
			}
		}
	}
}

// Observe durable transitions, not timing/polling of an unsynchronized row.
type notifyCommandStore struct {
	commandStore
	changed func()
}

func (s *notifyCommandStore) NextWorkAt(ctx context.Context, scope Scope) (time.Time, bool, error) {
	return s.commandStore.(commandScheduleStore).NextWorkAt(ctx, scope)
}
func (s *notifyCommandStore) Retry(ctx context.Context, claim Claim, next time.Time, code string) (Command, error) {
	command, err := s.commandStore.Retry(ctx, claim, next, code)
	if err == nil {
		s.changed()
	}
	return command, err
}
func (s *notifyCommandStore) IsolatePostDispatch(ctx context.Context, claim Claim, status int, remote, code string) (Command, error) {
	command, err := s.commandStore.(postDispatchIsolationStore).IsolatePostDispatch(ctx, claim, status, remote, code)
	if err == nil {
		s.changed()
	}
	return command, err
}

func TestServiceYieldsFullAcceptanceBatchToProviderAfterReadiness(t *testing.T) {
	scope := outboundIntentTestScope()
	source := &saturatedIntentSource{revision: scope.CatalogRevision}
	acceptStore := &acceptAllStore{}
	acceptor, _ := NewAcceptor(source, acceptStore, scope)
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	started := false
	store := &memoryCommandStore{command: testWorkerCommand()}
	observed := &notifyCommandStore{commandStore: store, changed: func() {
		if !started || source.lists != 1 || acceptStore.accepted != outboundintentv1.MaxIntentsPerList {
			t.Errorf("not a bounded post-readiness turn: ready=%t lists=%d accepted=%d", started, source.lists, acceptStore.accepted)
		}
		cancel()
	}}
	worker, _ := NewWorker(observed, failingPreparer{err: &PreparationError{Code: "association_link_pending", Retryable: true, Cause: ErrDependency}}, &scriptedCommitter{}, scope, "worker")
	service, _ := NewService(acceptor, worker)
	err := service.Run(ctx, func() { started = true })
	if !errors.Is(err, context.Canceled) || store.command.State != StateRetry {
		t.Fatalf("run=%v state=%s", err, store.command.State)
	}
}

type recoveringIntentSource struct {
	scriptedIntentSource
	operation     string
	failAt        int
	calls         int
	err           error
	recovered     chan struct{}
	signaled      bool
	closed        bool
	beforeFailure func()
}

func (s *recoveringIntentSource) fail(op string) error {
	if op != s.operation {
		return nil
	}
	s.calls++
	if s.calls == s.failAt {
		if s.beforeFailure != nil {
			s.beforeFailure()
		}
		return s.err
	}
	if s.calls > s.failAt {
		s.signaled = true
	}
	return nil
}
func (s *recoveringIntentSource) ReadOutboundIntentGeneration(ctx context.Context, r outboundintentv1.GenerationRequest) (outboundintentv1.GenerationResponse, error) {
	if err := s.fail("generation"); err != nil {
		return outboundintentv1.GenerationResponse{}, err
	}
	response, err := s.scriptedIntentSource.ReadOutboundIntentGeneration(ctx, r)
	if s.signaled && s.accepted && !s.closed {
		s.closed = true
		close(s.recovered)
	}
	return response, err
}
func (s *recoveringIntentSource) ListPendingOutboundIntents(ctx context.Context, r outboundintentv1.ListRequest) (outboundintentv1.ListResponse, error) {
	if err := s.fail("list"); err != nil {
		return outboundintentv1.ListResponse{}, err
	}
	return s.scriptedIntentSource.ListPendingOutboundIntents(ctx, r)
}
func (s *recoveringIntentSource) AcknowledgeOutboundIntent(ctx context.Context, r outboundintentv1.AcknowledgeRequest) (outboundintentv1.AcknowledgeResponse, error) {
	if err := s.fail("ack"); err != nil {
		return outboundintentv1.AcknowledgeResponse{}, err
	}
	return s.scriptedIntentSource.AcknowledgeOutboundIntent(ctx, r)
}

func TestServiceRecoversTransientRuntimeFailuresWithoutRestart(t *testing.T) {
	for _, tc := range []struct {
		name, operation string
		failAt          int
		category        commonv1.FailureCategory
	}{
		{"startup_generation", "generation", 1, commonv1.FailureUnavailable},
		{"steady_generation", "generation", 2, commonv1.FailureUnavailable},
		{"list_deadline", "list", 1, commonv1.FailureDeadlineExceeded},
		{"ack_resource_pressure", "ack", 1, commonv1.FailureResourceExhausted},
	} {
		t.Run(tc.name, func(t *testing.T) {
			scope := outboundIntentTestScope()
			source := &recoveringIntentSource{scriptedIntentSource: scriptedIntentSource{intent: outboundIntentTestIntent(scope.CatalogRevision, "recover", "record")}, operation: tc.operation, failAt: tc.failAt, err: &commonv1.Failure{Category: tc.category}, recovered: make(chan struct{})}
			store := &memoryAcceptStore{}
			acceptor, _ := NewAcceptor(source, store, scope)
			worker, _ := NewWorker(&memoryCommandStore{}, &countingPreparer{}, &scriptedCommitter{}, scope, "worker")
			service, _ := NewService(acceptor, worker)
			service.generationProbeInterval = time.Millisecond
			service.recoveryReconcileInterval = time.Hour
			started := 0
			source.beforeFailure = func() {
				if tc.name == "startup_generation" && started != 0 {
					t.Error("ready before authority recovered")
				}
			}
			ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second)
			defer cancel()
			done := make(chan error, 1)
			go func() { done <- service.Run(ctx, func() { started++ }) }()
			select {
			case <-source.recovered:
			case err := <-done:
				t.Fatalf("owner exited instead of recovering: %v", err)
			case <-ctx.Done():
				t.Fatal("recovery timed out")
			}
			cancel()
			if err := <-done; !errors.Is(err, context.Canceled) {
				t.Fatal(err)
			}
			if started != 1 {
				t.Fatalf("startup callbacks=%d", started)
			}
			if tc.operation == "ack" && (store.calls != 2 || store.command.CommandID != "command-1") {
				t.Fatalf("lost ACK did not reuse durable acceptance: calls=%d command=%s", store.calls, store.command.CommandID)
			}
		})
	}
}

func TestServiceDoesNotRetryPermanentRuntimeFailure(t *testing.T) {
	scope := outboundIntentTestScope()
	source := &recoveringIntentSource{scriptedIntentSource: scriptedIntentSource{intent: outboundIntentTestIntent(scope.CatalogRevision, "bad", "record")}, operation: "generation", failAt: 1, err: errors.Join(ErrIdentityConflict, &commonv1.Failure{Category: commonv1.FailureUnavailable}), recovered: make(chan struct{})}
	acceptor, _ := NewAcceptor(source, &memoryAcceptStore{}, scope)
	worker, _ := NewWorker(&memoryCommandStore{}, &countingPreparer{}, &scriptedCommitter{}, scope, "worker")
	service, _ := NewService(acceptor, worker)
	started := false
	err := service.Run(t.Context(), func() { started = true })
	if !errors.Is(err, ErrIdentityConflict) || started || source.calls != 1 {
		t.Fatalf("err=%v started=%t calls=%d", err, started, source.calls)
	}
}