package runtimeevents

import (
	"context"
	"strings"
	"sync"
	"testing"
	"time"
)

func TestPublishFanOut(t *testing.T) {
	b := NewBus(WithRingSize(16))
	defer b.Close()

	s1 := b.Subscribe("s1", nil, 64, DropNewest)
	s2 := b.Subscribe("s2", nil, 64, DropNewest)

	b.Publish(context.Background(), NewEvent(KindSessionStarted))

	select {
	case got := <-s1.Events():
		if got.Kind != KindSessionStarted {
			t.Errorf("s1 got kind %q, want %q", got.Kind, KindSessionStarted)
		}
	case <-time.After(time.Second):
		t.Error("s1 timed out waiting for event")
	}

	select {
	case got := <-s2.Events():
		if got.Kind != KindSessionStarted {
			t.Errorf("s2 got kind %q, want %q", got.Kind, KindSessionStarted)
		}
	case <-time.After(time.Second):
		t.Error("s2 timed out waiting for event")
	}
}

func TestFilterSubscriber(t *testing.T) {
	b := NewBus(WithRingSize(16))
	defer b.Close()

	wakeOnly := FilterKind(KindWakeReceived, KindWakeReplySent)
	s1 := b.Subscribe("wake-sink", wakeOnly, 64, DropNewest)

	b.Publish(context.Background(), NewEvent(KindSessionStarted))
	b.Publish(context.Background(), NewEvent(KindWakeReceived))

	select {
	case got := <-s1.Events():
		if got.Kind != KindWakeReceived {
			t.Errorf("got kind %q, want %q", got.Kind, KindWakeReceived)
		}
	case <-time.After(time.Second):
		t.Error("timed out")
	}

	// SessionStarted should not have been delivered.
	select {
	case got := <-s1.Events():
		t.Errorf("unexpected event: %q", got.Kind)
	default:
	}
}

func TestDropNewest(t *testing.T) {
	b := NewBus(WithRingSize(16))
	defer b.Close()

	s := b.Subscribe("tiny", nil, 2, DropNewest)

	for i := 0; i < 5; i++ {
		b.Publish(context.Background(), NewEvent(KindTurnThinking))
	}

	count := 0
	for {
		select {
		case <-s.Events():
			count++
		default:
			goto done
		}
	}
done:
	if count != 2 {
		t.Errorf("received %d events, want 2", count)
	}
	if dropped := s.Dropped(); dropped != 3 {
		t.Errorf("dropped = %d, want 3", dropped)
	}
}

func TestDropOldest(t *testing.T) {
	b := NewBus(WithRingSize(16))
	defer b.Close()

	s := b.Subscribe("tiny", nil, 2, DropOldest)

	var ids []string
	for i := 0; i < 5; i++ {
		ev := NewEvent(KindTurnText)
		b.Publish(context.Background(), ev)
		ids = append(ids, ev.ID)
	}

	var received []string
	for {
		select {
		case ev := <-s.Events():
			received = append(received, ev.ID)
		default:
			goto done2
		}
	}
done2:
	if len(received) != 2 {
		t.Fatalf("received %d events, want 2", len(received))
	}
	if received[0] != ids[3] {
		t.Errorf("first received = %s, want %s", received[0], ids[3])
	}
	if received[1] != ids[4] {
		t.Errorf("second received = %s, want %s", received[1], ids[4])
	}
}

func TestRecent(t *testing.T) {
	b := NewBus(WithRingSize(4))
	defer b.Close()

	for i := 0; i < 6; i++ {
		b.Publish(context.Background(), NewEvent(KindSessionStarted))
	}

	all := b.Recent(nil, 10)
	if len(all) != 4 {
		t.Errorf("Recent() returned %d events, want 4", len(all))
	}
}

func TestRecentStoresScrubbedEvents(t *testing.T) {
	b := NewBus(WithRingSize(16))
	defer b.Close()

	sub := b.Subscribe("metrics-like", nil, 1, DropNewest)
	b.Publish(context.Background(), NewDataEvent(NewUsageTokensData("sk-secret-123", 1, 2, 3, 0)))

	select {
	case ev := <-sub.Events():
		data, ok := DataAs[*UsageTokensData](ev)
		if !ok {
			t.Fatal("DataAs[*UsageTokensData] failed for subscriber event")
		}
		if data.APIKey != "sk-secret-123" {
			t.Fatalf("subscriber APIKey = %q, want original", data.APIKey)
		}
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for subscriber event")
	}

	recent := b.Recent(nil, 10)
	if len(recent) != 1 {
		t.Fatalf("Recent() returned %d events, want 1", len(recent))
	}
	data, ok := DataAs[*UsageTokensData](recent[0])
	if !ok {
		t.Fatal("DataAs[*UsageTokensData] failed for recent event")
	}
	if data.APIKey != "***" {
		t.Fatalf("recent APIKey = %q, want scrubbed", data.APIKey)
	}
}

func TestRecentRingByteLimitEvictsOldest(t *testing.T) {
	sample := ScrubEvent(NewDataEvent(NewTurnTextData(KindTurnText, strings.Repeat("x", 600))))
	limit := estimateEventBytes(sample) * 3
	b := NewBus(WithRingSize(16), WithRingByteLimit(limit))
	defer b.Close()

	var lastID string
	for i := 0; i < 10; i++ {
		ev := NewDataEvent(NewTurnTextData(KindTurnText, strings.Repeat("x", 600)))
		lastID = ev.ID
		b.Publish(context.Background(), ev)
	}

	recent := b.Recent(nil, 20)
	if len(recent) >= 10 {
		t.Fatalf("Recent() retained %d events, want byte-limit eviction", len(recent))
	}
	if got := recent[len(recent)-1].ID; got != lastID {
		t.Fatalf("last recent ID = %q, want latest %q", got, lastID)
	}
}

func TestRecentWithFilter(t *testing.T) {
	b := NewBus(WithRingSize(16))
	defer b.Close()

	b.Publish(context.Background(), NewEvent(KindSessionStarted))
	b.Publish(context.Background(), NewEvent(KindWakeReceived))
	b.Publish(context.Background(), NewEvent(KindSessionStopped))
	b.Publish(context.Background(), NewEvent(KindWakeReplySent))

	wakes := b.Recent(FilterKindPrefix("wake."), 10)
	if len(wakes) != 2 {
		t.Errorf("Recent(wake) returned %d, want 2", len(wakes))
	}
}

func TestCloseClosesChannels(t *testing.T) {
	b := NewBus(WithRingSize(16))
	s := b.Subscribe("s", nil, 64, DropNewest)
	b.Close()

	_, ok := <-s.Events()
	if ok {
		t.Error("expected channel to be closed after Close()")
	}
}

func TestPublishAfterClose(t *testing.T) {
	b := NewBus(WithRingSize(16))
	b.Close()

	// Should not panic.
	b.Publish(context.Background(), NewEvent(KindSessionStarted))
}

func TestCloseDuringPublishDoesNotPanic(t *testing.T) {
	for attempt := 0; attempt < 20; attempt++ {
		b := NewBus(WithRingSize(16))
		for i := 0; i < 16; i++ {
			b.Subscribe("s"+string(rune('a'+i)), nil, 1, DropOldest)
		}

		start := make(chan struct{})
		var wg sync.WaitGroup
		for i := 0; i < 16; i++ {
			wg.Add(1)
			go func() {
				defer wg.Done()
				<-start
				for j := 0; j < 100; j++ {
					b.Publish(context.Background(), NewEvent(KindSessionStarted))
				}
			}()
		}
		close(start)
		b.Close()
		wg.Wait()
	}
}

func TestSubscribeAfterCloseReturnsClosedSubscription(t *testing.T) {
	b := NewBus(WithRingSize(16))
	b.Close()

	s := b.Subscribe("late", nil, 1, DropNewest)
	select {
	case _, ok := <-s.Events():
		if ok {
			t.Fatal("Subscribe after Close returned open events channel")
		}
	case <-time.After(time.Second):
		t.Fatal("Subscribe after Close returned a channel that did not close")
	}
}

func TestSinkPanicIsolation(t *testing.T) {
	b := NewBus(WithRingSize(16))
	defer b.Close()

	s := b.Subscribe("panicker", nil, 64, DropNewest)
	go func() {
		for range s.Events() {
			return
		}
	}()

	s2 := b.Subscribe("stable", nil, 64, DropNewest)

	b.Publish(context.Background(), NewEvent(KindSessionStarted))

	select {
	case <-s2.Events():
	case <-time.After(time.Second):
		t.Error("s2 should receive first event")
	}

	b.Publish(context.Background(), NewEvent(KindWakeReceived))

	select {
	case ev := <-s2.Events():
		if ev.Kind != KindWakeReceived {
			t.Errorf("s2 got %q, want %q", ev.Kind, KindWakeReceived)
		}
	case <-time.After(time.Second):
		t.Error("s2 should still receive events after s1 goroutine exits")
	}
}

func TestNewDataEvent(t *testing.T) {
	data := NewTurnTextData(KindTurnText, "hello world")
	ev := NewDataEvent(data)

	if ev.Kind != KindTurnText {
		t.Errorf("Kind = %q, want %q", ev.Kind, KindTurnText)
	}
	if d, ok := DataAs[*TurnTextData](ev); !ok {
		t.Error("DataAs[*TurnTextData] failed")
	} else if d.Text != "hello world" {
		t.Errorf("Text = %q, want %q", d.Text, "hello world")
	}
}

func TestPublisherWithData(t *testing.T) {
	b := NewBus(WithRingSize(16))
	defer b.Close()

	s := b.Subscribe("s", nil, 64, DropNewest)

	pub := NewPublisher(b, EventContext{AgentID: "agent1"}, EventTrace{})
	pub.PublishData(context.Background(), NewWakeData(KindWakeReceived, "test"),
		WithMessageID("msg1"),
	)

	select {
	case ev := <-s.Events():
		if ev.Kind != KindWakeReceived {
			t.Errorf("Kind = %q, want %q", ev.Kind, KindWakeReceived)
		}
		if ev.Context.AgentID != "agent1" {
			t.Errorf("AgentID = %q, want %q", ev.Context.AgentID, "agent1")
		}
		if ev.Trace.MessageID != "msg1" {
			t.Errorf("MessageID = %q, want %q", ev.Trace.MessageID, "msg1")
		}
		if d, ok := DataAs[*WakeData](ev); !ok {
			t.Error("DataAs[*WakeData] failed")
		} else if d.Detail != "test" {
			t.Errorf("Detail = %q, want %q", d.Detail, "test")
		}
	case <-time.After(time.Second):
		t.Error("timed out")
	}
}