package proto

import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"math"
	"math/rand"
	"strconv"
	"strings"
	"testing"
	"time"
)

// segmentedReader delivers its payload one configured chunk per Read, so a bufio
// reader over it observes the exact partial-buffer boundaries we want to test
// (e.g. a push-notification header split across socket reads).
type segmentedReader struct {
	chunks [][]byte
}

func (s *segmentedReader) Read(p []byte) (int, error) {
	for len(s.chunks) > 0 && len(s.chunks[0]) == 0 {
		s.chunks = s.chunks[1:]
	}
	if len(s.chunks) == 0 {
		return 0, io.EOF
	}
	n := copy(p, s.chunks[0])
	s.chunks[0] = s.chunks[0][n:]
	return n, nil
}

// TestPeekPushNotificationName_LongName verifies that a notification name
// longer than the historical 36-byte window still parses.
func TestPeekPushNotificationName_LongName(t *testing.T) {
	longName := strings.Repeat("A", 100)
	buf := createValidPushNotification(longName, "data")
	r := createReaderWithPrimedBuffer(buf)
	name, err := r.PeekPushNotificationName()
	if err != nil {
		t.Fatalf("long name should parse, got error: %v", err)
	}
	if name != longName {
		t.Fatalf("expected %d-char name, got %q", len(longName), name)
	}
}

// TestPeekPushNotificationName_ExceedsBuffer verifies that when the header does
// not fit the reader's buffer, PeekPushNotificationName reports
// ErrPushNotificationNameTooLong (so the caller consumes the frame) instead of
// returning bufio.ErrBufferFull or stalling.
func TestPeekPushNotificationName_ExceedsBuffer(t *testing.T) {
	// invalidate header (">2\r\n$10\r\ninvalidate\r\n") is ~21 bytes; use a 16-byte
	// buffer so it cannot be peeked in full.
	full := ">2\r\n$10\r\ninvalidate\r\n$1\r\nk\r\n"
	r := NewReaderSize(bytes.NewReader([]byte(full)), 16)
	if _, err := r.Peek(1); err != nil {
		t.Fatalf("prime peek: %v", err)
	}
	_, err := r.PeekPushNotificationName()
	if !errors.Is(err, ErrPushNotificationNameTooLong) {
		t.Fatalf("expected ErrPushNotificationNameTooLong, got %v", err)
	}
	// The frame is still fully consumable via ReadReply (how the push processor
	// makes progress in this case).
	reply, err := r.ReadReply()
	if err != nil {
		t.Fatalf("ReadReply after too-long peek: %v", err)
	}
	if arr, ok := reply.([]interface{}); !ok || len(arr) != 2 || arr[0] != "invalidate" {
		t.Fatalf("unexpected reply: %#v", reply)
	}
}

// TestPeekPushNotificationName_SplitHeader reproduces a header split across reads:
// the first read yields only part of the "$<len>\r\n<name>\r\n" name line. The
// old implementation clamped its peek to the buffered bytes and never refilled,
// so it returned a parse error (stalling repeated drain passes on the same bytes).
// The fix grows the peek window, forcing a refill, so the name resolves.
func TestPeekPushNotificationName_SplitHeader(t *testing.T) {
	full := []byte(">2\r\n$10\r\ninvalidate\r\n$1\r\nk\r\n")
	for _, split := range []int{1, 5, 9, 14, 19} { // cut before/inside the name line
		t.Run(fmt.Sprintf("split_%d", split), func(t *testing.T) {
			r := NewReader(&segmentedReader{chunks: [][]byte{full[:split], full[split:]}})
			if _, err := r.Peek(1); err != nil {
				t.Fatalf("prime peek: %v", err)
			}
			name, err := r.PeekPushNotificationName()
			if err != nil {
				t.Fatalf("split header must resolve, got error: %v", err)
			}
			if name != "invalidate" {
				t.Fatalf("expected 'invalidate', got %q", name)
			}
			// Peek must not consume: the full frame is still readable.
			reply, err := r.ReadReply()
			if err != nil {
				t.Fatalf("ReadReply after peek: %v", err)
			}
			arr, ok := reply.([]interface{})
			if !ok || len(arr) != 2 || arr[0] != "invalidate" {
				t.Fatalf("unexpected reply after split-header peek: %#v", reply)
			}
		})
	}
}

// TestPeekPushNotificationName tests the updated PeekPushNotificationName method
func TestPeekPushNotificationName(t *testing.T) {
	t.Run("ValidPushNotifications", func(t *testing.T) {
		testCases := []struct {
			name         string
			notification string
			expected     string
		}{
			{"MOVING", "MOVING", "MOVING"},
			{"MIGRATING", "MIGRATING", "MIGRATING"},
			{"MIGRATED", "MIGRATED", "MIGRATED"},
			{"FAILING_OVER", "FAILING_OVER", "FAILING_OVER"},
			{"FAILED_OVER", "FAILED_OVER", "FAILED_OVER"},
			{"message", "message", "message"},
			{"pmessage", "pmessage", "pmessage"},
			{"subscribe", "subscribe", "subscribe"},
			{"unsubscribe", "unsubscribe", "unsubscribe"},
			{"psubscribe", "psubscribe", "psubscribe"},
			{"punsubscribe", "punsubscribe", "punsubscribe"},
			{"smessage", "smessage", "smessage"},
			{"ssubscribe", "ssubscribe", "ssubscribe"},
			{"sunsubscribe", "sunsubscribe", "sunsubscribe"},
			{"custom", "custom", "custom"},
			{"short", "a", "a"},
			{"empty", "", ""},
		}

		for _, tc := range testCases {
			t.Run(tc.name, func(t *testing.T) {
				buf := createValidPushNotification(tc.notification, "data")
				reader := NewReader(buf)

				// Prime the buffer by peeking first
				_, _ = reader.rd.Peek(1)

				name, err := reader.PeekPushNotificationName()
				if err != nil {
					t.Errorf("PeekPushNotificationName should not error for valid notification: %v", err)
				}

				if name != tc.expected {
					t.Errorf("Expected notification name '%s', got '%s'", tc.expected, name)
				}
			})
		}
	})

	t.Run("NotificationWithMultipleArguments", func(t *testing.T) {
		// Create push notification with multiple arguments
		buf := createPushNotificationWithArgs("MOVING", "slot", "123", "from", "node1", "to", "node2")
		reader := NewReader(buf)

		// Prime the buffer
		_, _ = reader.rd.Peek(1)

		name, err := reader.PeekPushNotificationName()
		if err != nil {
			t.Errorf("PeekPushNotificationName should not error: %v", err)
		}

		if name != "MOVING" {
			t.Errorf("Expected 'MOVING', got '%s'", name)
		}
	})

	t.Run("SingleElementNotification", func(t *testing.T) {
		// Create push notification with single element
		buf := createSingleElementPushNotification("TEST")
		reader := NewReader(buf)

		// Prime the buffer
		_, _ = reader.rd.Peek(1)

		name, err := reader.PeekPushNotificationName()
		if err != nil {
			t.Errorf("PeekPushNotificationName should not error: %v", err)
		}

		if name != "TEST" {
			t.Errorf("Expected 'TEST', got '%s'", name)
		}
	})

	t.Run("ErrorDetection", func(t *testing.T) {
		t.Run("NotPushNotification", func(t *testing.T) {
			// Test with regular array instead of push notification
			buf := &bytes.Buffer{}
			fmt.Fprint(buf, "*2\r\n$6\r\nMOVING\r\n$4\r\ndata\r\n")
			reader := NewReader(buf)

			_, err := reader.PeekPushNotificationName()
			if err == nil {
				t.Error("PeekPushNotificationName should error for non-push notification")
			}

			// The error might be "no data available" or "can't parse push notification"
			if !strings.Contains(err.Error(), "can't peek push notification name") {
				t.Errorf("Error should mention push notification parsing, got: %v", err)
			}
		})

		t.Run("InsufficientData", func(t *testing.T) {
			// Test with buffer smaller than peek size - this might panic due to bounds checking
			buf := &bytes.Buffer{}
			fmt.Fprint(buf, ">")
			reader := NewReader(buf)

			func() {
				defer func() {
					if r := recover(); r != nil {
						t.Logf("PeekPushNotificationName panicked as expected for insufficient data: %v", r)
					}
				}()
				_, err := reader.PeekPushNotificationName()
				if err == nil {
					t.Error("PeekPushNotificationName should error for insufficient data")
				}
			}()
		})

		t.Run("EmptyBuffer", func(t *testing.T) {
			buf := &bytes.Buffer{}
			reader := NewReader(buf)

			_, err := reader.PeekPushNotificationName()
			if err == nil {
				t.Error("PeekPushNotificationName should error for empty buffer")
			}
		})

		t.Run("DifferentRESPTypes", func(t *testing.T) {
			// Test with different RESP types that should be rejected
			respTypes := []byte{'+', '-', ':', '$', '*', '%', '~', '|', '('}

			for _, respType := range respTypes {
				t.Run(fmt.Sprintf("Type_%c", respType), func(t *testing.T) {
					buf := &bytes.Buffer{}
					buf.WriteByte(respType)
					fmt.Fprint(buf, "test data that fills the buffer completely")
					reader := NewReader(buf)

					_, err := reader.PeekPushNotificationName()
					if err == nil {
						t.Errorf("PeekPushNotificationName should error for RESP type '%c'", respType)
					}

					// The error might be "no data available" or "can't parse push notification"
					if !strings.Contains(err.Error(), "can't peek push notification name") {
						t.Errorf("Error should mention push notification parsing, got: %v", err)
					}
				})
			}
		})
	})

	t.Run("EdgeCases", func(t *testing.T) {
		t.Run("ZeroLengthArray", func(t *testing.T) {
			// Create push notification with zero elements: >0\r\n
			buf := &bytes.Buffer{}
			fmt.Fprint(buf, ">0\r\npadding_data_to_fill_buffer_completely")
			reader := NewReader(buf)

			_, err := reader.PeekPushNotificationName()
			if err == nil {
				t.Error("PeekPushNotificationName should error for zero-length array")
			}
		})

		t.Run("EmptyNotificationName", func(t *testing.T) {
			// Create push notification with empty name: >1\r\n$0\r\n\r\n
			buf := createValidPushNotification("", "data")
			reader := NewReader(buf)

			// Prime the buffer
			_, _ = reader.rd.Peek(1)

			name, err := reader.PeekPushNotificationName()
			if err != nil {
				t.Errorf("PeekPushNotificationName should not error for empty name: %v", err)
			}

			if name != "" {
				t.Errorf("Expected empty notification name, got '%s'", name)
			}
		})

		t.Run("CorruptedData", func(t *testing.T) {
			corruptedCases := []struct {
				name string
				data string
			}{
				{"CorruptedLength", ">abc\r\n$6\r\nMOVING\r\n"},
				{"MissingCRLF", ">2$6\r\nMOVING\r\n$4\r\ndata\r\n"},
				{"InvalidStringLength", ">2\r\n$abc\r\nMOVING\r\n$4\r\ndata\r\n"},
				{"NegativeStringLength", ">2\r\n$-1\r\n$4\r\ndata\r\n"},
				{"IncompleteString", ">1\r\n$6\r\nMOV"},
			}

			for _, tc := range corruptedCases {
				t.Run(tc.name, func(t *testing.T) {
					buf := &bytes.Buffer{}
					fmt.Fprint(buf, tc.data)
					reader := NewReader(buf)

					// Some corrupted data might not error but return unexpected results
					// This is acceptable behavior for malformed input
					name, err := reader.PeekPushNotificationName()
					if err != nil {
						t.Logf("PeekPushNotificationName errored for corrupted data %s: %v (DATA: %s)", tc.name, err, tc.data)
					} else {
						t.Logf("PeekPushNotificationName returned '%s' for corrupted data NAME: %s, DATA: %s", name, tc.name, tc.data)
					}
				})
			}
		})
	})

	t.Run("BoundaryConditions", func(t *testing.T) {
		t.Run("ExactlyPeekSize", func(t *testing.T) {
			// Create buffer that is exactly 36 bytes (the peek window size)
			buf := &bytes.Buffer{}
			// ">1\r\n$4\r\nTEST\r\n" = 14 bytes, need 22 more
			fmt.Fprint(buf, ">1\r\n$4\r\nTEST\r\n1234567890123456789012")
			if buf.Len() != 36 {
				t.Errorf("Expected buffer length 36, got %d", buf.Len())
			}

			reader := NewReader(buf)
			// Prime the buffer
			_, _ = reader.rd.Peek(1)

			name, err := reader.PeekPushNotificationName()
			if err != nil {
				t.Errorf("PeekPushNotificationName should work for exact peek size: %v", err)
			}

			if name != "TEST" {
				t.Errorf("Expected 'TEST', got '%s'", name)
			}
		})

		t.Run("LessThanPeekSize", func(t *testing.T) {
			// Create buffer smaller than 36 bytes but with complete notification
			buf := createValidPushNotification("TEST", "")
			reader := NewReader(buf)

			// Prime the buffer
			_, _ = reader.rd.Peek(1)

			name, err := reader.PeekPushNotificationName()
			if err != nil {
				t.Errorf("PeekPushNotificationName should work for complete notification: %v", err)
			}

			if name != "TEST" {
				t.Errorf("Expected 'TEST', got '%s'", name)
			}
		})

		t.Run("LongNotificationName", func(t *testing.T) {
			// Test with notification name that might exceed peek window
			longName := strings.Repeat("A", 20) // 20 character name (safe size)
			buf := createValidPushNotification(longName, "data")
			reader := NewReader(buf)

			// Prime the buffer
			_, _ = reader.rd.Peek(1)

			name, err := reader.PeekPushNotificationName()
			if err != nil {
				t.Errorf("PeekPushNotificationName should work for long name: %v", err)
			}

			if name != longName {
				t.Errorf("Expected '%s', got '%s'", longName, name)
			}
		})
	})
}

// Helper functions to create test data

// createValidPushNotification creates a valid RESP3 push notification
func createValidPushNotification(notificationName, data string) *bytes.Buffer {
	buf := &bytes.Buffer{}

	simpleOrString := rand.Intn(2) == 0
	defMsg := fmt.Sprintf("$%d\r\n%s\r\n", len(notificationName), notificationName)

	if data == "" {

		// Single element notification
		fmt.Fprint(buf, ">1\r\n")
		if simpleOrString {
			fmt.Fprintf(buf, "+%s\r\n", notificationName)
		} else {
			fmt.Fprint(buf, defMsg)
		}
	} else {
		// Two element notification
		fmt.Fprint(buf, ">2\r\n")
		if simpleOrString {
			fmt.Fprintf(buf, "+%s\r\n", notificationName)
			fmt.Fprintf(buf, "+%s\r\n", data)
		} else {
			fmt.Fprint(buf, defMsg)
			fmt.Fprint(buf, defMsg)
		}
	}

	return buf
}

// createReaderWithPrimedBuffer creates a reader and primes the buffer
func createReaderWithPrimedBuffer(buf *bytes.Buffer) *Reader {
	reader := NewReader(buf)
	// Prime the buffer by peeking first
	_, _ = reader.rd.Peek(1)
	return reader
}

// createPushNotificationWithArgs creates a push notification with multiple arguments
func createPushNotificationWithArgs(notificationName string, args ...string) *bytes.Buffer {
	buf := &bytes.Buffer{}

	totalElements := 1 + len(args)
	fmt.Fprintf(buf, ">%d\r\n", totalElements)

	// Write notification name
	fmt.Fprintf(buf, "$%d\r\n%s\r\n", len(notificationName), notificationName)

	// Write arguments
	for _, arg := range args {
		fmt.Fprintf(buf, "$%d\r\n%s\r\n", len(arg), arg)
	}

	return buf
}

// createSingleElementPushNotification creates a push notification with single element
func createSingleElementPushNotification(notificationName string) *bytes.Buffer {
	buf := &bytes.Buffer{}
	fmt.Fprint(buf, ">1\r\n")
	fmt.Fprintf(buf, "$%d\r\n%s\r\n", len(notificationName), notificationName)
	return buf
}

// BenchmarkPeekPushNotificationName benchmarks the method performance
func BenchmarkPeekPushNotificationName(b *testing.B) {
	testCases := []struct {
		name         string
		notification string
	}{
		{"Short", "TEST"},
		{"Medium", "MOVING_NOTIFICATION"},
		{"Long", "VERY_LONG_NOTIFICATION_NAME_FOR_TESTING"},
	}

	for _, tc := range testCases {
		b.Run(tc.name, func(b *testing.B) {
			buf := createValidPushNotification(tc.notification, "data")
			data := buf.Bytes()

			// Reuse both bytes.Reader and proto.Reader to avoid allocations
			bytesReader := bytes.NewReader(data)
			reader := NewReader(bytesReader)

			b.ResetTimer()
			b.ReportAllocs()
			for i := 0; i < b.N; i++ {
				// Reset the bytes.Reader to the beginning without allocating
				bytesReader.Reset(data)
				// Reset the proto.Reader to reuse the bufio buffer
				reader.Reset(bytesReader)
				_, err := reader.PeekPushNotificationName()
				if err != nil {
					b.Errorf("PeekPushNotificationName should not error: %v", err)
				}
			}
		})
	}
}

// TestPeekPushNotificationNameSpecialCases tests special cases and realistic scenarios
func TestPeekPushNotificationNameSpecialCases(t *testing.T) {
	t.Run("RealisticNotifications", func(t *testing.T) {
		// Test realistic Redis push notifications
		realisticCases := []struct {
			name         string
			notification []string
			expected     string
		}{
			{"MovingSlot", []string{"MOVING", "slot", "123", "from", "127.0.0.1:7000", "to", "127.0.0.1:7001"}, "MOVING"},
			{"MigratingSlot", []string{"MIGRATING", "slot", "456", "from", "127.0.0.1:7001", "to", "127.0.0.1:7002"}, "MIGRATING"},
			{"MigratedSlot", []string{"MIGRATED", "slot", "789", "from", "127.0.0.1:7002", "to", "127.0.0.1:7000"}, "MIGRATED"},
			{"FailingOver", []string{"FAILING_OVER", "node", "127.0.0.1:7000"}, "FAILING_OVER"},
			{"FailedOver", []string{"FAILED_OVER", "node", "127.0.0.1:7000"}, "FAILED_OVER"},
			{"PubSubMessage", []string{"message", "mychannel", "hello world"}, "message"},
			{"PubSubPMessage", []string{"pmessage", "pattern*", "mychannel", "hello world"}, "pmessage"},
			{"Subscribe", []string{"subscribe", "mychannel", "1"}, "subscribe"},
			{"Unsubscribe", []string{"unsubscribe", "mychannel", "0"}, "unsubscribe"},
		}

		for _, tc := range realisticCases {
			t.Run(tc.name, func(t *testing.T) {
				buf := createPushNotificationWithArgs(tc.notification[0], tc.notification[1:]...)
				reader := createReaderWithPrimedBuffer(buf)

				name, err := reader.PeekPushNotificationName()
				if err != nil {
					t.Errorf("PeekPushNotificationName should not error for %s: %v", tc.name, err)
				}

				if name != tc.expected {
					t.Errorf("Expected '%s', got '%s'", tc.expected, name)
				}
			})
		}
	})

	t.Run("SpecialCharactersInName", func(t *testing.T) {
		specialCases := []struct {
			name         string
			notification string
		}{
			{"WithUnderscore", "test_notification"},
			{"WithDash", "test-notification"},
			{"WithNumbers", "test123"},
			{"WithDots", "test.notification"},
			{"WithColon", "test:notification"},
			{"WithSlash", "test/notification"},
			{"MixedCase", "TestNotification"},
			{"AllCaps", "TESTNOTIFICATION"},
			{"AllLower", "testnotification"},
			{"Unicode", "tëst"},
		}

		for _, tc := range specialCases {
			t.Run(tc.name, func(t *testing.T) {
				buf := createValidPushNotification(tc.notification, "data")
				reader := createReaderWithPrimedBuffer(buf)

				name, err := reader.PeekPushNotificationName()
				if err != nil {
					t.Errorf("PeekPushNotificationName should not error for '%s': %v", tc.notification, err)
				}

				if name != tc.notification {
					t.Errorf("Expected '%s', got '%s'", tc.notification, name)
				}
			})
		}
	})

	t.Run("IdempotentPeek", func(t *testing.T) {
		// Test that multiple peeks return the same result
		buf := createValidPushNotification("MOVING", "data")
		reader := createReaderWithPrimedBuffer(buf)

		// First peek
		name1, err1 := reader.PeekPushNotificationName()
		if err1 != nil {
			t.Errorf("First PeekPushNotificationName should not error: %v", err1)
		}

		// Second peek should return the same result
		name2, err2 := reader.PeekPushNotificationName()
		if err2 != nil {
			t.Errorf("Second PeekPushNotificationName should not error: %v", err2)
		}

		if name1 != name2 {
			t.Errorf("Peek should be idempotent: first='%s', second='%s'", name1, name2)
		}

		if name1 != "MOVING" {
			t.Errorf("Expected 'MOVING', got '%s'", name1)
		}
	})
}

// TestPeekPushNotificationNamePerformance tests performance characteristics
func TestPeekPushNotificationNamePerformance(t *testing.T) {
	t.Run("RepeatedCalls", func(t *testing.T) {
		// Test that repeated calls work correctly
		buf := createValidPushNotification("TEST", "data")
		reader := createReaderWithPrimedBuffer(buf)

		// Call multiple times
		for i := 0; i < 10; i++ {
			name, err := reader.PeekPushNotificationName()
			if err != nil {
				t.Errorf("PeekPushNotificationName should not error on call %d: %v", i, err)
			}
			if name != "TEST" {
				t.Errorf("Expected 'TEST' on call %d, got '%s'", i, name)
			}
		}
	})

	t.Run("LargeNotifications", func(t *testing.T) {
		// Test with large notification data
		largeData := strings.Repeat("x", 1000)
		buf := createValidPushNotification("LARGE", largeData)
		reader := createReaderWithPrimedBuffer(buf)

		name, err := reader.PeekPushNotificationName()
		if err != nil {
			t.Errorf("PeekPushNotificationName should not error for large notification: %v", err)
		}

		if name != "LARGE" {
			t.Errorf("Expected 'LARGE', got '%s'", name)
		}
	})
}

// TestPeekPushNotificationNameBehavior documents the method's behavior
func TestPeekPushNotificationNameBehavior(t *testing.T) {
	t.Run("MethodBehavior", func(t *testing.T) {
		// Test that the method works as intended:
		// 1. Peek at the buffer without consuming it
		// 2. Detect push notifications (RESP type '>')
		// 3. Extract the notification name from the first element
		// 4. Return the name for filtering decisions

		buf := createValidPushNotification("MOVING", "slot_data")
		reader := createReaderWithPrimedBuffer(buf)

		// Peek should not consume the buffer
		name, err := reader.PeekPushNotificationName()
		if err != nil {
			t.Errorf("PeekPushNotificationName should not error: %v", err)
		}

		if name != "MOVING" {
			t.Errorf("Expected 'MOVING', got '%s'", name)
		}

		// Buffer should still be available for normal reading
		replyType, err := reader.PeekReplyType()
		if err != nil {
			t.Errorf("PeekReplyType should work after PeekPushNotificationName: %v", err)
		}

		if replyType != RespPush {
			t.Errorf("Expected RespPush, got %v", replyType)
		}
	})

	t.Run("BufferNotConsumed", func(t *testing.T) {
		// Verify that peeking doesn't consume the buffer
		buf := createValidPushNotification("TEST", "data")
		originalData := buf.Bytes()
		reader := createReaderWithPrimedBuffer(buf)

		// Peek the notification name
		name, err := reader.PeekPushNotificationName()
		if err != nil {
			t.Errorf("PeekPushNotificationName should not error: %v", err)
		}

		if name != "TEST" {
			t.Errorf("Expected 'TEST', got '%s'", name)
		}

		// Read the actual notification
		reply, err := reader.ReadReply()
		if err != nil {
			t.Errorf("ReadReply should work after peek: %v", err)
		}

		// Verify we got the complete notification
		if replySlice, ok := reply.([]interface{}); ok {
			if len(replySlice) != 2 {
				t.Errorf("Expected 2 elements, got %d", len(replySlice))
			}
			if replySlice[0] != "TEST" {
				t.Errorf("Expected 'TEST', got %v", replySlice[0])
			}
		} else {
			t.Errorf("Expected slice reply, got %T", reply)
		}

		// Verify buffer was properly consumed
		if buf.Len() != 0 {
			t.Errorf("Buffer should be empty after reading, but has %d bytes: %q", buf.Len(), buf.Bytes())
		}

		t.Logf("Original buffer size: %d bytes", len(originalData))
		t.Logf("Successfully peeked and then read complete notification")
	})

	t.Run("ImplementationSuccess", func(t *testing.T) {
		// Document that the implementation is now working correctly
		t.Log("PeekPushNotificationName implementation status:")
		t.Log("1. ✅ Correctly parses RESP3 push notifications")
		t.Log("2. ✅ Extracts notification names properly")
		t.Log("3. ✅ Handles buffer peeking without consumption")
		t.Log("4. ✅ Works with various notification types")
		t.Log("5. ✅ Supports empty notification names")
		t.Log("")
		t.Log("RESP3 format parsing:")
		t.Log(">2\\r\\n$6\\r\\nMOVING\\r\\n$4\\r\\ndata\\r\\n")
		t.Log("✅ Correctly identifies push notification marker (>)")
		t.Log("✅ Skips array length (2)")
		t.Log("✅ Parses string marker ($) and length (6)")
		t.Log("✅ Extracts notification name (MOVING)")
		t.Log("✅ Returns name without consuming buffer")
		t.Log("")
		t.Log("Note: Buffer must be primed with a peek operation first")
	})
}

// chunkedReader hands its data out in fixed-size chunks, one per Read. This
// lets us deterministically reproduce the bufio fill boundary that landed in
// the middle of a push frame header in issue #3839.
type chunkedReader struct {
	data      []byte
	chunkSize int
	off       int
}

func (r *chunkedReader) Read(p []byte) (int, error) {
	if r.off >= len(r.data) {
		return 0, io.EOF
	}
	n := r.chunkSize
	if n > len(p) {
		n = len(p)
	}
	if r.off+n > len(r.data) {
		n = len(r.data) - r.off
	}
	copy(p, r.data[r.off:r.off+n])
	r.off += n
	return n, nil
}

// TestPeekPushNotificationName_TruncatedHeader covers issue #3839: when only
// a prefix of a push frame is buffered, PeekPushNotificationName must NOT
// return a truncated name. It should block to fetch the rest of the header
// (so the caller can identify the notification correctly) or, when the
// underlying read fails, return that error.
func TestPeekPushNotificationName_TruncatedHeader(t *testing.T) {
	const frame = ">3\r\n$7\r\nmessage\r\n$7\r\nchannel\r\n$5\r\nhello\r\n"

	// Boundaries that previously caused PeekPushNotificationName to silently
	// return a truncated name. `prefix` is the number of bytes the chunked
	// reader hands out per Read call; with the bug, every chunk boundary that
	// lands inside the frame header is a place where the function returned a
	// truncated name instead of blocking for the rest.
	cutpoints := []struct {
		name   string
		prefix int
	}{
		{"after_push_marker", 1},
		{"inside_array_len_line", 3},
		{"after_array_len_line", 4},
		{"after_dollar", 5},
		{"inside_bulk_len_line", 6},
		{"after_bulk_len_line", 8},
		{"inside_name", 13}, // the exact failure mode from #3839 ("messa")
		{"before_name_crlf", 15},
	}

	for _, tc := range cutpoints {
		t.Run(tc.name, func(t *testing.T) {
			data := []byte(frame)
			// Use a chunked reader that hands out tc.prefix bytes per Read
			// call (not per fill). bufio's first fill therefore returns only
			// tc.prefix bytes; satisfying a Peek that wants more forces
			// additional Reads, each capped at tc.prefix bytes, until the
			// peek window is full. That sequence reproduces the partial-fill
			// boundary from issue #3839 where the buffered prefix landed in
			// the middle of the push frame header.
			rd := NewReader(&chunkedReader{data: data, chunkSize: tc.prefix})

			if _, err := rd.PeekReplyType(); err != nil {
				t.Fatalf("PeekReplyType: %v", err)
			}

			name, err := rd.PeekPushNotificationName()
			if err != nil {
				t.Fatalf("PeekPushNotificationName: unexpected error: %v", err)
			}
			if name != "message" {
				t.Fatalf("PeekPushNotificationName: got %q, want %q", name, "message")
			}

			// And the full frame must still be readable afterwards.
			reply, err := rd.ReadReply()
			if err != nil {
				t.Fatalf("ReadReply: %v", err)
			}
			arr, ok := reply.([]interface{})
			if !ok || len(arr) != 3 || arr[0] != "message" || arr[1] != "channel" || arr[2] != "hello" {
				t.Fatalf("ReadReply: got %#v, want [message channel hello]", reply)
			}
		})
	}
}

// TestPeekPushNotificationName_TruncatedHeaderUnderlyingEOF covers the case
// where the underlying connection delivers a partial frame and then closes.
// PeekPushNotificationName must surface the read error rather than returning
// a truncated name.
func TestPeekPushNotificationName_TruncatedHeaderUnderlyingEOF(t *testing.T) {
	// A push frame that ends mid-name.
	data := []byte(">3\r\n$7\r\nmessa")
	rd := NewReader(bytes.NewReader(data))

	if _, err := rd.PeekReplyType(); err != nil {
		t.Fatalf("PeekReplyType: %v", err)
	}

	name, err := rd.PeekPushNotificationName()
	if err == nil {
		t.Fatalf("PeekPushNotificationName: want error, got name=%q", name)
	}
	if name != "" {
		t.Fatalf("PeekPushNotificationName: want empty name on error, got %q", name)
	}
}

// TestPeekPushNotificationName_EmptyArrayLength asserts that ">\r\n" is
// reported as a malformed frame rather than an incomplete prefix. RESP
// requires at least one digit after '>' for the array length; without the
// empty-length check in parsePushNotificationName the parser would treat
// ">\r\n" as a valid prefix and PeekPushNotificationName would block
// waiting for the rest of a frame that is already corrupt.
func TestPeekPushNotificationName_EmptyArrayLength(t *testing.T) {
	rd := NewReader(bytes.NewReader([]byte(">\r\n")))

	if _, err := rd.PeekReplyType(); err != nil {
		t.Fatalf("PeekReplyType: %v", err)
	}

	name, err := rd.PeekPushNotificationName()
	if err == nil {
		t.Fatalf("PeekPushNotificationName: want error for empty array length, got name=%q", name)
	}
	if name != "" {
		t.Fatalf("PeekPushNotificationName: want empty name on error, got %q", name)
	}
	if !strings.Contains(err.Error(), "empty push notification array length") {
		t.Fatalf("PeekPushNotificationName: error should mention empty array length, got %v", err)
	}
}

// TestPeekPushNotificationName_OverflowingNameLength asserts that a frame
// advertising a name length near math.MaxInt does not panic. Computing
// next+nameLen would overflow int and wrap negative, slipping past a naive
// "end > len(buf)" guard and panicking the backing slice; the parser must
// instead treat it as incomplete and surface an error without crashing.
func TestPeekPushNotificationName_OverflowingNameLength(t *testing.T) {
	data := []byte(">1\r\n$" + strconv.Itoa(math.MaxInt) + "\r\n")
	rd := NewReader(bytes.NewReader(data))

	if _, err := rd.PeekReplyType(); err != nil {
		t.Fatalf("PeekReplyType: %v", err)
	}

	name, err := rd.PeekPushNotificationName()
	if err == nil {
		t.Fatalf("PeekPushNotificationName: want error for overflowing name length, got name=%q", name)
	}
	if name != "" {
		t.Fatalf("PeekPushNotificationName: want empty name on error, got %q", name)
	}
}

// idleConnReader serves scripted chunks and then blocks on the next Read, the
// way an idle network connection does. Unlike the bytes/strings readers used
// elsewhere in this file it never returns io.EOF on its own, so a Peek that
// asks for more bytes than the server will ever send blocks instead of
// erroring — the situation of issue #3935.
type idleConnReader struct {
	chunks   [][]byte
	released chan struct{} // closed by the test to unblock a stuck Read
}

func (r *idleConnReader) Read(p []byte) (int, error) {
	if len(r.chunks) == 0 {
		<-r.released
		return 0, io.EOF
	}
	chunk := r.chunks[0]
	n := copy(p, chunk)
	if n == len(chunk) {
		r.chunks = r.chunks[1:]
	} else {
		r.chunks[0] = chunk[n:]
	}
	return n, nil
}

// peekNameWithTimeout runs PeekPushNotificationName in a goroutine and fails
// the test if it does not return within the deadline, which is how the hang
// of issue #3935 manifests.
func peekNameWithTimeout(t *testing.T, rd *Reader) (string, error) {
	t.Helper()

	type result struct {
		name string
		err  error
	}
	done := make(chan result, 1)
	go func() {
		name, err := rd.PeekPushNotificationName()
		done <- result{name, err}
	}()

	select {
	case res := <-done:
		return res.name, res.err
	case <-time.After(5 * time.Second):
		t.Fatal("PeekPushNotificationName blocked waiting for bytes the server will not send")
		return "", nil
	}
}

// TestPeekPushNotificationName_ShortFrameIdleConnection reproduces issue
// #3935: the subscribe confirmation for a channel name of six or fewer
// characters is a complete push frame shorter than 36 bytes, the fixed
// initial peek window, and with the connection otherwise idle
// PeekPushNotificationName blocked forever inside bufio.Reader.Peek waiting
// for bytes the server had no reason to send. The name must be identified
// from the bytes actually available.
func TestPeekPushNotificationName_ShortFrameIdleConnection(t *testing.T) {
	testCases := []struct {
		name  string
		frame string
		want  string
	}{
		// 35 bytes, one short of the old fixed 36-byte peek window.
		{"subscribe six-char channel", ">3\r\n$9\r\nsubscribe\r\n$6\r\naaaaaa\r\n:1\r\n", "subscribe"},
		{"subscribe one-char channel", ">3\r\n$9\r\nsubscribe\r\n$1\r\na\r\n:1\r\n", "subscribe"},
		{"message with short payload", ">3\r\n$7\r\nmessage\r\n$1\r\na\r\n$1\r\nb\r\n", "message"},
		{"minimal push frame", ">1\r\n$1\r\nx\r\n", "x"},
		{"simple-string name", ">2\r\n+MOVING\r\n:1\r\n", "MOVING"},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			src := &idleConnReader{
				chunks:   [][]byte{[]byte(tc.frame)},
				released: make(chan struct{}),
			}
			defer close(src.released)
			rd := NewReader(src)

			if _, err := rd.PeekReplyType(); err != nil {
				t.Fatalf("PeekReplyType: %v", err)
			}

			name, err := peekNameWithTimeout(t, rd)
			if err != nil {
				t.Fatalf("PeekPushNotificationName: %v", err)
			}
			if name != tc.want {
				t.Fatalf("PeekPushNotificationName: got %q, want %q", name, tc.want)
			}
		})
	}
}

// TestPeekPushNotificationName_ByteByByteDelivery feeds the frame one byte
// per Read, mimicking worst-case TCP fragmentation. The peek loop must keep
// fetching single bytes until the name is parseable and never demand more
// than the next byte.
func TestPeekPushNotificationName_ByteByByteDelivery(t *testing.T) {
	frame := ">3\r\n$9\r\nsubscribe\r\n$6\r\naaaaaa\r\n:1\r\n"
	chunks := make([][]byte, 0, len(frame))
	for i := 0; i < len(frame); i++ {
		chunks = append(chunks, []byte{frame[i]})
	}
	src := &idleConnReader{chunks: chunks, released: make(chan struct{})}
	defer close(src.released)
	rd := NewReader(src)

	if _, err := rd.PeekReplyType(); err != nil {
		t.Fatalf("PeekReplyType: %v", err)
	}

	name, err := peekNameWithTimeout(t, rd)
	if err != nil {
		t.Fatalf("PeekPushNotificationName: %v", err)
	}
	if name != "subscribe" {
		t.Fatalf("PeekPushNotificationName: got %q, want %q", name, "subscribe")
	}
}