package event

import (
	"context"
	"io"
	"log/slog"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

func testLogger() *slog.Logger {
	return slog.New(slog.NewTextHandler(io.Discard, nil))
}

// waitFor koşulu sağlanana kadar bekler (dağıtım asenkron olduğu için gerekli).
func waitFor(t *testing.T, cond func() bool, msg string) {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(2 * time.Millisecond)
	}
	t.Fatalf(msg)
}

// Olay türü abonelerine dağıtılmalı.
func TestPublish_DeliversToTypeSubscribers(t *testing.T) {
	bus := NewMemoryBus(testLogger(), 16)
	defer bus.Close()

	got := make(chan Event, 1)
	bus.Subscribe(UserMessageReceived, func(_ context.Context, e Event) { got <- e })

	bus.Publish(context.Background(), Event{Type: UserMessageReceived, UserID: "u1", ConversationID: "c1"})

	select {
	case e := <-got:
		if e.UserID != "u1" || e.ConversationID != "c1" {
			t.Fatalf("olay içeriği bozuk: %+v", e)
		}
		if e.At.IsZero() {
			t.Error("At damgası otomatik atanmalıydı")
		}
	case <-time.After(3 * time.Second):
		t.Fatal("olay teslim edilmedi")
	}
}

// Farklı türe abone olan işleyici ALMAMALI (filtreleme).
func TestPublish_DoesNotDeliverToOtherType(t *testing.T) {
	bus := NewMemoryBus(testLogger(), 16)
	defer bus.Close()

	var wrong atomic.Int32
	bus.Subscribe(AgentCompleted, func(context.Context, Event) { wrong.Add(1) })

	bus.Publish(context.Background(), Event{Type: UserMessageReceived})
	waitFor(t, func() bool { return wrong.Load() == 0 }, "istenmeyen olay teslim edildi")
	bus.Close()
}

// SubscribeAll her türü almalı.
func TestPublish_SubscribeAllGetsEverything(t *testing.T) {
	bus := NewMemoryBus(testLogger(), 16)
	defer bus.Close()

	got := make(chan Type, 4)
	bus.SubscribeAll(func(_ context.Context, e Event) { got <- e.Type })

	bus.Publish(context.Background(), Event{Type: UserMessageReceived})
	bus.Publish(context.Background(), Event{Type: AgentStarted})
	bus.Publish(context.Background(), Event{Type: AgentCompleted})

	seen := map[Type]bool{}
	for i := 0; i < 3; i++ {
		select {
		case tp := <-got:
			seen[tp] = true
		case <-time.After(3 * time.Second):
			t.Fatal("SubscribeAll olayları almadı")
		}
	}
	if len(seen) != 3 {
		t.Errorf("3 farklı tür bekleniyordu: %v", seen)
	}
}

// Abonelik kaldırıldıktan sonra olay gelmemeli (bellek sızıntısı olmasın).
func TestUnsubscribe_StopsDelivery(t *testing.T) {
	bus := NewMemoryBus(testLogger(), 16)
	defer bus.Close()

	var n atomic.Int32
	unsub := bus.Subscribe(AgentStarted, func(context.Context, Event) { n.Add(1) })

	bus.Publish(context.Background(), Event{Type: AgentStarted})
	waitFor(t, func() bool { return n.Load() == 1 }, "ilk olay teslim edilmedi")

	unsub()
	bus.Publish(context.Background(), Event{Type: AgentStarted})
	time.Sleep(50 * time.Millisecond)

	if n.Load() != 1 {
		t.Errorf("abonelik kaldırıldıktan sonra olay gelmemeliydi, sayaç=%d", n.Load())
	}
}

// Yayıncı ASLA bloklanmamalı: kuyruk dolarsa olay düşer.
func TestPublish_NonBlockingWhenQueueFull(t *testing.T) {
	bus := NewMemoryBus(testLogger(), 1)
	defer bus.Close()

	// Hiç abone yok → dağıtıcı boşta; kuyruk doldurulur.
	done := make(chan struct{})
	go func() {
		defer close(done)
		for i := 0; i < 100; i++ {
			bus.Publish(context.Background(), Event{Type: UserMessageReceived})
		}
	}()

	select {
	case <-done:
	case <-time.After(3 * time.Second):
		t.Fatal("Publish blokladı — sözleşme ihlali")
	}
}

// Kapanmış bus'ta yayın yapmak çökmemeli, sessizce yok sayılmalı.
func TestPublish_AfterCloseIsSafe(t *testing.T) {
	bus := NewMemoryBus(testLogger(), 4)
	bus.Close()
	bus.Close() // ikinci çağrı da güvenli olmalı

	bus.Publish(context.Background(), Event{Type: AgentStarted}) // çökmemeli
}

// Close kuyruktaki olayları boşaltmalı (dışarıda bırakılmamalı).
func TestClose_DrainsQueuedEvents(t *testing.T) {
	bus := NewMemoryBus(testLogger(), 64)

	var n atomic.Int32
	bus.SubscribeAll(func(context.Context, Event) { n.Add(1) })

	for i := 0; i < 20; i++ {
		bus.Publish(context.Background(), Event{Type: AgentStarted})
	}
	bus.Close() // kapanana kadar bekler

	if n.Load() != 20 {
		t.Errorf("kapanmadan önce 20 olay işlenmeliydi, işlenen=%d", n.Load())
	}
}

// Bir işleyici panik yaparsa dağıtım devam etmeli (tek abone süreci düşürmemeli).
func TestHandlerPanic_DoesNotStopBus(t *testing.T) {
	bus := NewMemoryBus(testLogger(), 16)
	defer bus.Close()

	var ok atomic.Int32
	bus.Subscribe(AgentStarted, func(context.Context, Event) { panic("bilerek") })
	bus.Subscribe(AgentStarted, func(context.Context, Event) { ok.Add(1) })

	bus.Publish(context.Background(), Event{Type: AgentStarted})
	waitFor(t, func() bool { return ok.Load() == 1 }, "panik diğer işleyicileri engelledi")

	// Bus hâlâ çalışıyor olmalı.
	bus.Publish(context.Background(), Event{Type: AgentStarted})
	waitFor(t, func() bool { return ok.Load() == 2 }, "panik sonrası bus çalışmıyor")
}

// Eşzamanlı yayın/abonelik güvenli olmalı (veri yarışı olmamalı).
func TestConcurrentPublishSubscribe(t *testing.T) {
	bus := NewMemoryBus(testLogger(), 256)
	defer bus.Close()

	var received atomic.Int32
	bus.SubscribeAll(func(context.Context, Event) { received.Add(1) })

	var wg sync.WaitGroup
	for i := 0; i < 4; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			for j := 0; j < 50; j++ {
				unsub := bus.Subscribe(AgentStarted, func(context.Context, Event) {})
				bus.Publish(context.Background(), Event{Type: AgentStarted, Data: map[string]any{"i": i, "j": j}})
				unsub()
			}
		}(i)
	}
	wg.Wait()
	waitFor(t, func() bool { return received.Load() == 200 }, "tüm olaylar işlenmeliydi")
}