package mpubsub

import (
	"context"
	. "testing"
	"time"

	"github.com/mediocregopher/mediocre-go-lib/mcfg"
	"github.com/mediocregopher/mediocre-go-lib/mdb"
	"github.com/mediocregopher/mediocre-go-lib/mrand"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

var testPS *PubSub

func init() {
	mdb.DefaultGCEProject = "test"
	cfg := mcfg.New()
	testPS = Cfg(cfg)
	cfg.StartTestRun()
}

// this requires the pubsub emulator to be running
func TestPubSub(t *T) {
	topicName := "testTopic_" + mrand.Hex(8)
	ctx := context.Background()

	// Topic shouldn't exist yet
	_, err := testPS.Topic(ctx, topicName, false)
	require.Error(t, err)

	// ...so create it
	topic, err := testPS.Topic(ctx, topicName, true)
	require.NoError(t, err)

	// Create a subscription and consumer
	sub, err := topic.Subscription(ctx, "testSub", true)
	require.NoError(t, err)

	msgCh := make(chan *Message)
	go sub.Consume(ctx, func(ctx context.Context, m *Message) (bool, error) {
		msgCh <- m
		return true, nil
	}, ConsumerOpts{})
	time.Sleep(1 * time.Second) // give consumer time to actually start

	// publish a message and make sure it gets consumed
	assert.NoError(t, topic.Publish(ctx, []byte("foo")))
	msg := <-msgCh
	assert.Equal(t, []byte("foo"), msg.Data)
}

func TestBatchPubSub(t *T) {
	ctx := context.Background()
	topicName := "testBatchTopic_" + mrand.Hex(8)
	topic, err := testPS.Topic(ctx, topicName, true)
	require.NoError(t, err)

	readBatch := func(ch chan []*Message) map[byte]int {
		select {
		case <-time.After(1 * time.Second):
			assert.Fail(t, "waited too long to read batch")
			return nil
		case mm := <-ch:
			ret := map[byte]int{}
			for _, m := range mm {
				ret[m.Data[0]]++
			}
			return ret
		}
	}

	// we use the same sub across the next two sections to ensure that cleanup
	// also works
	sub, err := topic.Subscription(ctx, "testSub", true)
	require.NoError(t, err)
	sub.batchTestTrigger = make(chan bool)

	{ // no grouping
		// Create a subscription and consumer
		ctx, cancel := context.WithCancel(ctx)
		ch := make(chan []*Message)
		go func() {
			sub.BatchConsume(ctx,
				func(ctx context.Context, mm []*Message) (bool, error) {
					ch <- mm
					return true, nil
				},
				nil,
				ConsumerOpts{Concurrent: 5},
			)
			close(ch)
		}()
		time.Sleep(1 * time.Second) // give consumer time to actually start

		exp := map[byte]int{}
		for i := byte(0); i <= 9; i++ {
			require.NoError(t, topic.Publish(ctx, []byte{i}))
			exp[i] = 1
		}

		time.Sleep(1 * time.Second)
		sub.batchTestTrigger <- true
		gotA := readBatch(ch)
		assert.Len(t, gotA, 5)

		time.Sleep(1 * time.Second)
		sub.batchTestTrigger <- true
		gotB := readBatch(ch)
		assert.Len(t, gotB, 5)

		for i, c := range gotB {
			gotA[i] += c
		}
		assert.Equal(t, exp, gotA)

		time.Sleep(1 * time.Second) // give time to ack before cancelling
		cancel()
		<-ch
	}

	{ // with grouping
		ctx, cancel := context.WithCancel(ctx)
		ch := make(chan []*Message)
		go func() {
			sub.BatchConsume(ctx,
				func(ctx context.Context, mm []*Message) (bool, error) {
					ch <- mm
					return true, nil
				},
				func(a, b *Message) bool { return a.Data[0]%2 == b.Data[0]%2 },
				ConsumerOpts{Concurrent: 10},
			)
			close(ch)
		}()
		time.Sleep(1 * time.Second) // give consumer time to actually start

		exp := map[byte]int{}
		for i := byte(0); i <= 9; i++ {
			require.NoError(t, topic.Publish(ctx, []byte{i}))
			exp[i] = 1
		}

		time.Sleep(1 * time.Second)
		sub.batchTestTrigger <- true
		gotA := readBatch(ch)
		assert.Len(t, gotA, 5)
		gotB := readBatch(ch)
		assert.Len(t, gotB, 5)

		assertGotGrouped := func(got map[byte]int) {
			prev := byte(255)
			for i := range got {
				if prev != 255 {
					assert.Equal(t, prev%2, i%2)
				}
				prev = i
			}
		}

		assertGotGrouped(gotA)
		assertGotGrouped(gotB)
		for i, c := range gotB {
			gotA[i] += c
		}
		assert.Equal(t, exp, gotA)

		time.Sleep(1 * time.Second) // give time to ack before cancelling
		cancel()
		<-ch
	}
}