package connector_webhooks

import (
	"context"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	connectorcapabilities "github.com/pingcap-inc/nexuscrm-connectors/features/connector/capabilities"
	connector_inbound_routing "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
)

func TestMaterializerRouteCacheSingleflightsConcurrentMiss(t *testing.T) {
	resolver := &blockingRouteResolver{
		entered: make(chan struct{}),
		release: make(chan struct{}),
	}
	cache := newMaterializerRouteCache(time.Minute)

	const callers = 16
	var wg sync.WaitGroup
	start := make(chan struct{})
	errs := make(chan error, callers)
	for i := 0; i < callers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			_, _, err := cache.resolve(context.Background(), resolver, testNamespace, "install-1", "contact.created")
			errs <- err
		}()
	}
	close(start)
	waitForChannel(t, resolver.entered, "route resolver load")
	time.Sleep(25 * time.Millisecond)
	close(resolver.release)
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatalf("route cache resolve: %v", err)
		}
	}
	if got := resolver.calls.Load(); got != 1 {
		t.Fatalf("route resolver calls = %d, want 1 concurrent load", got)
	}
}

func TestMaterializerParserSingleflightCollapsesConcurrentResolve(t *testing.T) {
	parser := &fakeParser{
		events: []connectorcapabilities.ConnectorEvent{
			{EventType: "contact.created", Namespace: testNamespace, ObjectKind: "contact", SourceID: "src-a", Payload: []byte(`{}`), IdempotencyKey: "k0"},
		},
	}
	entered := make(chan struct{})
	release := make(chan struct{})
	var enteredOnce sync.Once
	var gateCalls atomic.Int32
	reg := connectorcapabilities.NewRegistry(func(context.Context, string, connectorcapabilities.CapabilityKind, string) (bool, error) {
		gateCalls.Add(1)
		enteredOnce.Do(func() { close(entered) })
		<-release
		return true, nil
	})
	if err := reg.UpsertCapabilityForPackInstall(testNamespace, connectorcapabilities.EnvelopeParser, "install-1", parser); err != nil {
		t.Fatalf("UpsertCapabilityForPackInstall: %v", err)
	}
	sf := newMaterializerParserSingleflight()

	const callers = 16
	var wg sync.WaitGroup
	start := make(chan struct{})
	errs := make(chan error, callers)
	for i := 0; i < callers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			_, err := sf.resolve(context.Background(), reg, testNamespace, "install-1")
			errs <- err
		}()
	}
	close(start)
	waitForChannel(t, entered, "parser capability load")
	time.Sleep(25 * time.Millisecond)
	close(release)
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatalf("parser cache resolve: %v", err)
		}
	}
	if got := gateCalls.Load(); got != 1 {
		t.Fatalf("parser capability gate calls = %d, want 1 concurrent load", got)
	}
}

func TestMaterializerCompanionSingleflightsConcurrentResolution(t *testing.T) {
	entered := make(chan struct{})
	release := make(chan struct{})
	var enteredOnce sync.Once
	var calls atomic.Int32
	d := &Dispatcher{
		resolveCompanion: func(context.Context, string, string) (CompanionFlowIncarnation, bool, error) {
			calls.Add(1)
			enteredOnce.Do(func() { close(entered) })
			<-release
			return CompanionFlowIncarnation{
				CompanionPackName:      "companion-app",
				CompanionPackInstallID: "companion-install",
				CompanionFlowVersionID: "flow-v1",
			}, true, nil
		},
		companionSingleflight: newMaterializerCompanionSingleflight(),
	}

	const callers = 16
	var wg sync.WaitGroup
	start := make(chan struct{})
	errs := make(chan error, callers)
	for i := 0; i < callers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			_, _, err := d.resolveMaterializerCompanion(context.Background(), testNamespace, "FlowA")
			errs <- err
		}()
	}
	close(start)
	waitForChannel(t, entered, "companion resolver load")
	time.Sleep(25 * time.Millisecond)
	close(release)
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatalf("resolveMaterializerCompanion: %v", err)
		}
	}
	if got := calls.Load(); got != 1 {
		t.Fatalf("companion resolver calls = %d, want 1 concurrent load", got)
	}
}

func TestMaterializerConsentSingleflightsConcurrentCheck(t *testing.T) {
	entered := make(chan struct{})
	release := make(chan struct{})
	var enteredOnce sync.Once
	var calls atomic.Int32
	d := &Dispatcher{
		checkConsent: func(context.Context, string, string, string, string) error {
			calls.Add(1)
			enteredOnce.Do(func() { close(entered) })
			<-release
			return nil
		},
		consentSingleflight: newMaterializerConsentSingleflight(),
	}

	const callers = 16
	var wg sync.WaitGroup
	start := make(chan struct{})
	errs := make(chan error, callers)
	for i := 0; i < callers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			errs <- d.checkMaterializerConsent(context.Background(), "companion-app", "companion-install", testNamespace, "v1")
		}()
	}
	close(start)
	waitForChannel(t, entered, "consent checker load")
	time.Sleep(25 * time.Millisecond)
	close(release)
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatalf("checkMaterializerConsent: %v", err)
		}
	}
	if got := calls.Load(); got != 1 {
		t.Fatalf("consent checker calls = %d, want 1 concurrent load", got)
	}
}

type blockingRouteResolver struct {
	calls       atomic.Int32
	entered     chan struct{}
	enteredOnce sync.Once
	release     chan struct{}
}

func (r *blockingRouteResolver) ResolveFlow(context.Context, string, string, string) (int64, string, error) {
	r.calls.Add(1)
	r.enteredOnce.Do(func() { close(r.entered) })
	<-r.release
	return 7, "FlowA", nil
}

func (r *blockingRouteResolver) ResolveFlowByVersion(context.Context, string, string, int64) (string, error) {
	return "FlowA", nil
}

func (r *blockingRouteResolver) ResolveExecutionSpecByVersion(ctx context.Context, installedPackID, eventKind string, version int64) (connector_inbound_routing.RouteExecutionSpec, error) {
	flow, err := r.ResolveFlowByVersion(ctx, installedPackID, eventKind, version)
	if err != nil {
		return connector_inbound_routing.RouteExecutionSpec{}, err
	}
	return connector_inbound_routing.NormalizeRouteExecutionSpec(connector_inbound_routing.RouteExecutionSpec{FlowDeveloperName: flow})
}

func waitForChannel(t *testing.T, ch <-chan struct{}, label string) {
	t.Helper()
	select {
	case <-ch:
	case <-time.After(time.Second):
		t.Fatalf("timed out waiting for %s", label)
	}
}