package server

import (
	"io"
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestRingBuffer(t *testing.T) {
	cases := []struct {
		name   string
		size   int
		writes [][]byte
		want   string
	}{
		{"PartialFill", 16, [][]byte{[]byte("hello")}, "hello"},
		{"ExactFill", 8, [][]byte{[]byte("01234567")}, "01234567"},
		{"WrapAround", 8, [][]byte{[]byte("01234567"), []byte("ABCD")}, "4567ABCD"}, // ring is "ABCD4567" but ordered "4567ABCD"
		{"LargerThanRing", 4, [][]byte{[]byte("0123456789")}, "6789"},               // only last 4 fit
		{"Empty", 16, [][]byte{{}}, ""},
		{"NoWrites", 16, nil, ""},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			r := newRingBuffer(tc.size)
			for _, w := range tc.writes {
				r.Write(w)
			}
			got := string(r.Snapshot())
			assert.Equal(t, tc.want, got)
		})
	}
}

func TestLogStream(t *testing.T) {
	t.Run("WriteBroadcastsToSubscriber", func(t *testing.T) {
		s := newLogStream()
		defer s.Close()
		sub, closed, _ := s.subscribe()
		assert.False(t, closed, "subscribe returned closed=true on a fresh stream")

		_, err := s.Write([]byte("line1\n"))
		assert.NoError(t, err)

		got, ok := <-sub.ch
		assert.True(t, ok, "channel closed before receiving")
		assert.Equal(t, "line1\n", string(got))
	})

	t.Run("SnapshotReturnsBackfill", func(t *testing.T) {
		s := newLogStream()
		defer s.Close()
		s.Write([]byte("early\n"))
		s.Write([]byte("later\n"))

		got := string(s.snapshot())
		assert.Equal(t, "early\nlater\n", got)
	})

	t.Run("WriteAfterClose", func(t *testing.T) {
		s := newLogStream()
		assert.NoError(t, s.Close())
		_, err := s.Write([]byte("x"))
		assert.ErrorIs(t, err, io.ErrClosedPipe)
	})

	t.Run("CloseSignalsSubscribers", func(t *testing.T) {
		s := newLogStream()
		sub, _, _ := s.subscribe()

		assert.NoError(t, s.Close())

		_, ok := <-sub.ch
		assert.False(t, ok, "channel still open after Close")

		// Idempotent.
		assert.NoError(t, s.Close(), "second Close")
	})

	t.Run("SubscribeAfterClose", func(t *testing.T) {
		s := newLogStream()
		s.Close()

		sub, closed, _ := s.subscribe()
		assert.True(t, closed, "subscribe after Close should return closed=true")
		_, ok := <-sub.ch
		assert.False(t, ok, "channel should be closed after subscribe-when-closed")
	})

	t.Run("DropsOnBackpressure", func(t *testing.T) {
		s := newLogStream()
		defer s.Close()
		sub, _, _ := s.subscribe()

		for range 200 {
			_, err := s.Write([]byte("x"))
			assert.NoError(t, err)
		}
		closed := false
		flushed := 0
		for !closed {
			select {
			case _, ok := <-sub.ch:
				if !ok {
					closed = true
				} else {
					flushed++
				}
			default:
				closed = true
			}
		}
		assert.NotZero(t, flushed, "expected at least one delivered write")
		assert.LessOrEqual(t, flushed, subChanDepth, "flushed %d exceeds channel depth", flushed)
	})

	t.Run("UnsubscribeRemovesFromMap", func(t *testing.T) {
		s := newLogStream()
		defer s.Close()
		sub, _, _ := s.subscribe()

		s.unsubscribe(sub)

		s.Write([]byte("x"))
		select {
		case got, ok := <-sub.ch:
			if ok {
				assert.Empty(t, got, "unexpected delivery after unsubscribe")
			}
		default:
		}
	})

	t.Run("MultipleSubscribers", func(t *testing.T) {
		s := newLogStream()
		defer s.Close()
		subA, _, _ := s.subscribe()
		subB, _, _ := s.subscribe()

		s.Write([]byte("hi"))

		for _, sub := range []<-chan []byte{subA.ch, subB.ch} {
			got := <-sub
			assert.Equal(t, "hi", string(got))
		}
	})

	t.Run("SubscribeLimit", func(t *testing.T) {
		s := newLogStream()
		for i := range maxSubscribersPerStream {
			_, _, err := s.subscribe()
			assert.NoError(t, err, "subscribe %d", i)
		}
		_, _, err := s.subscribe()
		assert.ErrorIs(t, err, errTooManySubscribers)
	})

	t.Run("UnsubscribeFreesSlot", func(t *testing.T) {
		s := newLogStream()
		sub, _, _ := s.subscribe()
		s.unsubscribe(sub)
		for i := range maxSubscribersPerStream {
			_, _, err := s.subscribe()
			assert.NoError(t, err, "subscribe %d after unsubscribe", i)
		}
	})
}

func TestLogHub(t *testing.T) {
	t.Run("GetReturnsNilOnFreshHub", func(t *testing.T) {
		h := newLogHub()
		assert.Nil(t, h.Get(), "Get on a fresh hub should return nil")
	})

	t.Run("GetOrCreateReusesOpenStream", func(t *testing.T) {
		h := newLogHub()
		s1 := h.GetOrCreate()
		s2 := h.GetOrCreate()
		assert.Same(t, s1, s2, "GetOrCreate should reuse the open stream")
	})

	t.Run("GetOrCreateRecreatesAfterClose", func(t *testing.T) {
		h := newLogHub()
		s1 := h.GetOrCreate()
		assert.NoError(t, s1.Close())
		s2 := h.GetOrCreate()
		assert.NotSame(t, s1, s2, "GetOrCreate should create a new stream after Close")
	})

	t.Run("GetReturnsCurrentStream", func(t *testing.T) {
		h := newLogHub()
		s := h.GetOrCreate()
		assert.Same(t, s, h.Get())
	})

	t.Run("CloseClosesStream", func(t *testing.T) {
		h := newLogHub()
		s := h.GetOrCreate()
		sub, _, _ := s.subscribe()

		h.Close()

		_, ok := <-sub.ch
		assert.False(t, ok, "subscriber channel should be closed after hub.Close")
	})

	t.Run("HubSinkCloseIsNoOp", func(t *testing.T) {
		h := newLogHub()
		sink := h.Stream("t1")
		assert.NoError(t, sink.Close())

		s := h.Get()
		assert.NotNil(t, s)
		_, err := s.Write([]byte("after-close"))
		assert.NoError(t, err, "underlying stream must still be open after hubSink.Close")
	})

	t.Run("StreamAggregatesAcrossNames", func(t *testing.T) {
		h := newLogHub()
		s1 := h.Stream("t1")
		_, err := s1.Write([]byte("tenant-a "))
		assert.NoError(t, err)
		s2 := h.Stream("t2")
		_, err = s2.Write([]byte("tenant-b"))
		assert.NoError(t, err)

		got := string(h.Get().snapshot())
		assert.Equal(t, "tenant-a tenant-b", got)
	})
}