package push

import (
	"context"
	"io"
	"strings"
	"testing"

	"github.com/redis/go-redis/v9/internal/proto"
)

// ========
// non-AI note: Test showcases a bug wherе a frame gets cut off in the
// notification name section:
// if frame is ">3\r\n$7\r\nmessage\r\n$7\r\nchannel\r\n$5\r\nhello\r\n" and
// on read it gets cutoff in the middle of the notification name, like
// ">3\r\n$7\r\nmessa" ProcessPendingNotifications will not break and will try
//  process the frame as a custom notification and drop it.
// ========

// chunkedReader hands out the underlying bytes in fixed-size chunks, one per
// Read call. This lets us deterministically reproduce the situation where
// bufio.Reader.fill() lands a chunk boundary in the middle of a RESP3 push
// frame, leaving only a partial frame (with a partial notification name) at
// the head of the buffer when PeekPushNotificationName is next called.
//
// bufio's fill() returns after the first non-empty Read, so the amount left
// buffered after a fill is exactly chunkSize.
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
}

// TestReproRESP3PubSubMessageLoss reproduces the go-redis v9 RESP3 pub/sub
// message-loss bug WITHOUT a real Redis server.
//
// It drives the exact code path that pubsub.go runs on every Receive():
//
//	rd.WithReader(func(rd) {
//	    processor.ProcessPendingNotifications(ctx, handlerCtx, rd) // drain out-of-band pushes
//	    cmd.readReply(rd)                                          // read the actual message
//	})
//
// A burst of identical RESP3 "message" push frames is fed through a chunked
// reader so that, at a chunk boundary, only the beginning of a "message"
// frame is buffered. PeekPushNotificationName then returns a TRUNCATED name
// (e.g. "messa") which is not in willHandleNotificationInClient's allow-list,
// so ProcessPendingNotifications consumes the frame with ReadReply and the
// message is silently dropped before the application ever sees it.
//
// On unpatched go-redis v9.19.0 this test FAILS: fewer messages are delivered
// than were published. That failure IS the reproduction of the bug.
func TestReproRESP3PubSubMessageLoss(t *testing.T) {
	// A single RESP3 pub/sub message push frame:
	//   >3\r\n            push array of 3 elements
	//   $7\r\nmessage\r\n the notification name
	//   $7\r\nchannel\r\n the channel
	//   $5\r\nhello\r\n   the payload
	// Total length: 41 bytes.
	const frame = ">3\r\n$7\r\nmessage\r\n$7\r\nchannel\r\n$5\r\nhello\r\n"
	const frameLen = len(frame)
	const numFrames = 1000

	// The publisher side: a pipelined MULTI/EXEC burst of 1000 PUBLISH
	// commands becomes 1000 push frames written to the subscriber socket
	// in one go.
	stream := strings.Repeat(frame, numFrames)

	// Choose a chunk size so that (chunkSize mod frameLen) lands in the
	// 9..16 byte range. After bufio consumes the whole-frames inside one
	// fill, that many leftover bytes form a partial "message" frame at the
	// head of the buffer - long enough to look like a push notification but
	// too short to contain the full name "message" + its terminating CRLF.
	//
	// frameLen = 41; 41*99 = 4059; 4072 - 4059 = 13 leftover bytes.
	chunkSize := 4072
	if rem := chunkSize % frameLen; rem < 9 || rem > 16 {
		t.Fatalf("test setup error: chunkSize %% frameLen = %d, want 9..16", rem)
	}

	rd := proto.NewReader(&chunkedReader{data: []byte(stream), chunkSize: chunkSize})
	processor := NewProcessor()
	ctx := context.Background()
	handlerCtx := NotificationHandlerContext{IsBlocking: true}

	delivered := 0
	for {
		// Step 1: drain pending out-of-band push notifications, exactly as
		// PubSub.ReceiveTimeout does before reading each reply.
		if err := processor.ProcessPendingNotifications(ctx, handlerCtx, rd); err != nil {
			t.Fatalf("ProcessPendingNotifications: %v", err)
		}

		// Step 2: read the actual pub/sub message (cmd.readReply).
		reply, err := rd.ReadReply()
		if err == io.EOF {
			break
		}
		if err != nil {
			t.Fatalf("ReadReply after %d delivered: %v", delivered, err)
		}

		arr, ok := reply.([]interface{})
		if !ok || len(arr) == 0 || arr[0] != "message" {
			t.Fatalf("unexpected reply after %d delivered: %#v", delivered, reply)
		}
		delivered++
	}

	if delivered != numFrames {
		t.Fatalf(
			"RESP3 pub/sub message loss reproduced: published %d messages but only %d were delivered (%d dropped)",
			numFrames, delivered, numFrames-delivered,
		)
	}
}