package streaming

import (
	"context"
	"reflect"
	"testing"

	"github.com/nexuscrm/kernel/pkg/ports"
	sharedEvents "github.com/nexuscrm/shared/pkg/events"
)

type bridgeTestEventBus struct {
	handlers map[sharedEvents.EventType][]ports.EventHandler
}

func newBridgeTestEventBus() *bridgeTestEventBus {
	return &bridgeTestEventBus{handlers: map[sharedEvents.EventType][]ports.EventHandler{}}
}

func (b *bridgeTestEventBus) Subscribe(eventType sharedEvents.EventType, handler ports.EventHandler) func() {
	b.handlers[eventType] = append(b.handlers[eventType], handler)
	return func() {
		handlers := b.handlers[eventType]
		for i, registered := range handlers {
			if reflect.ValueOf(registered).Pointer() == reflect.ValueOf(handler).Pointer() {
				b.handlers[eventType] = append(handlers[:i], handlers[i+1:]...)
				return
			}
		}
	}
}

func (b *bridgeTestEventBus) Publish(ctx context.Context, eventType sharedEvents.EventType, payload interface{}) error {
	for _, handler := range b.handlers[eventType] {
		if err := handler(ctx, payload); err != nil {
			return err
		}
	}
	return nil
}

func (b *bridgeTestEventBus) PublishAsyncWithContext(ctx context.Context, eventType sharedEvents.EventType, payload interface{}) {
	_ = b.Publish(ctx, eventType, payload)
}

type bridgeTestBroadcaster struct {
	broadcasts []bridgeTestBroadcast
}

type bridgeTestBroadcast struct {
	channel string
	data    interface{}
}

func (b *bridgeTestBroadcaster) Broadcast(channel string, data interface{}) {
	b.broadcasts = append(b.broadcasts, bridgeTestBroadcast{channel: channel, data: data})
}

func TestEventBridgeHasNoSalesforceTopicsWithoutRegistration(t *testing.T) {
	eventBus := newBridgeTestEventBus()
	broadcaster := &bridgeTestBroadcaster{}
	bridge := NewEventBridge(eventBus, broadcaster)

	bridge.Start()
	defer bridge.Stop()

	if got := len(eventBus.handlers); got != 0 {
		t.Fatalf("default L0 EventBridge subscribed to %d event type(s); want no topic subscriptions without product registration", got)
	}
	if err := eventBus.Publish(context.Background(), sharedEvents.RecordCreated, map[string]any{"id": "001"}); err != nil {
		t.Fatalf("Publish: %v", err)
	}
	if got := len(broadcaster.broadcasts); got != 0 {
		t.Fatalf("default L0 EventBridge broadcast %d message(s); want no Salesforce /topic output without product registration", got)
	}
}

func TestEventBridgeBroadcastsRegisteredTopic(t *testing.T) {
	eventBus := newBridgeTestEventBus()
	broadcaster := &bridgeTestBroadcaster{}
	bridge := NewEventBridge(eventBus, broadcaster)
	bridge.RegisterEventChannelMappings(EventChannelMapping{
		EventType: sharedEvents.RecordCreated,
		Channel:   "/topic/records/created",
	})

	bridge.Start()
	defer bridge.Stop()

	payload := map[string]any{"id": "001"}
	if err := eventBus.Publish(context.Background(), sharedEvents.RecordCreated, payload); err != nil {
		t.Fatalf("Publish: %v", err)
	}
	if got := len(broadcaster.broadcasts); got != 1 {
		t.Fatalf("broadcast count = %d, want 1", got)
	}
	if got := broadcaster.broadcasts[0].channel; got != "/topic/records/created" {
		t.Fatalf("broadcast channel = %q, want registered topic", got)
	}
	if !reflect.DeepEqual(broadcaster.broadcasts[0].data, payload) {
		t.Fatalf("broadcast payload = %#v, want %#v", broadcaster.broadcasts[0].data, payload)
	}
}

// TestEventBridgeMappingFilterGatesBroadcast pins the Filter seam: a mapping
// filter that returns false must drop the event from every broadcaster while
// a true verdict broadcasts as before. The bridge stays product-blind — the
// filter's policy comes entirely from the registration.
func TestEventBridgeMappingFilterGatesBroadcast(t *testing.T) {
	eventBus := newBridgeTestEventBus()
	broadcaster := &bridgeTestBroadcaster{}
	bridge := NewEventBridge(eventBus, broadcaster)
	filterCalls := 0
	bridge.RegisterEventChannelMappings(EventChannelMapping{
		EventType: sharedEvents.RecordCreated,
		Channel:   "/topic/records/created",
		Filter: func(_ context.Context, payload interface{}) bool {
			filterCalls++
			m, ok := payload.(map[string]any)
			return ok && m["keep"] == true
		},
	})

	bridge.Start()
	defer bridge.Stop()

	if err := eventBus.Publish(context.Background(), sharedEvents.RecordCreated, map[string]any{"keep": false}); err != nil {
		t.Fatalf("Publish(drop): %v", err)
	}
	if got := len(broadcaster.broadcasts); got != 0 {
		t.Fatalf("broadcast count after filtered event = %d, want 0", got)
	}

	kept := map[string]any{"keep": true}
	if err := eventBus.Publish(context.Background(), sharedEvents.RecordCreated, kept); err != nil {
		t.Fatalf("Publish(keep): %v", err)
	}
	if got := len(broadcaster.broadcasts); got != 1 {
		t.Fatalf("broadcast count after passing event = %d, want 1", got)
	}
	if !reflect.DeepEqual(broadcaster.broadcasts[0].data, kept) {
		t.Fatalf("broadcast payload = %#v, want %#v", broadcaster.broadcasts[0].data, kept)
	}
	if filterCalls != 2 {
		t.Fatalf("filter consulted %d time(s), want 2", filterCalls)
	}
}