package pubsubclient

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"net/http"
	"sync/atomic"
	"testing"
	"time"

	"github.com/gorilla/websocket"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

const (
	pubsubURL  = "wss://pubsub-edge-darklaunch.twitch.tv:443/v1"
	pubsterURL = "https://pubster-darklaunch.twitch.tv/publish"
)

func TestIntegration(t *testing.T) {
	assert := assert.New(t)

	c := New(pubsubURL, nil)
	// This topic name is special: darklaunch allows topics that start
	// with 'pubsubtest.'
	const topic = "pubsubtest.topic"
	assert.NoError(c.Subscribe(topic, ""))

	var sent []*Message
	var recvd []*Message
	const n = 10
	for i := 0; i < n; i++ {
		msgBody := fmt.Sprintf("msg-%d", i)
		m := &Message{Topic: topic, Message: msgBody}
		sent = append(sent, m)

		err := publishMessage(pubsterURL, m)
		require.NoError(t, err)

		msg, err := getMessage(c, 5*time.Second)

		require.NoError(t, err)
		recvd = append(recvd, msg)
	}

	assert.Equal(n, len(recvd))
	assert.Equal(sent, recvd)

	assert.NoError(c.Close())
}

func TestHardReconnectIntegration(t *testing.T) {
	// Use a dialer that lets us grab a reference to the network
	// connection
	var netconn net.Conn
	dialer := &websocket.Dialer{
		NetDial: func(network, addr string) (net.Conn, error) {
			var err error
			netconn, err = net.Dial(network, addr)
			return netconn, err
		},
	}

	// connected is a channel to signal when the newConn function is
	// executed.
	connected := make(chan struct{}, 1)
	newConn := func() (pubsubConn, error) {
		socket, _, err := dialer.Dial(pubsubURL, nil)
		if err != nil {
			return nil, err
		}
		conn := newPubsubConn(socket)
		select {
		case connected <- struct{}{}:
		default:
		}
		return conn, nil
	}

	c := newClient(newConn)

	// This topic name is special: darklaunch allows topics that start
	// with 'pubsubtest.'
	const topic = "pubsubtest.topic2"
	assert.NoError(t, c.Subscribe(topic, ""))
	select {
	case <-connected:
	case <-time.After(10 * time.Second):
		require.Fail(t, "timed out while waiting to connect")
	}

	var sent []*Message
	var recvd []*Message
	const n = 10
	for i := 0; i < n; i++ {
		msgBody := fmt.Sprintf("msg-%d", i)
		m := &Message{Topic: topic, Message: msgBody}
		sent = append(sent, m)

		if i == 5 {
			// Brutally break the connection!
			require.NoError(t, netconn.Close())
			// Wait until we reconnect.
			select {
			case <-connected:
			case <-time.After(10 * time.Second):
				require.Fail(t, "timed out while waiting to reconnect")
			}
			// There's a bit more work to do. Poll the internal state of the
			// client until the new connection is fully set up.
			deadline := time.Now().Add(5 * time.Second)
			for {
				if time.Now().After(deadline) {
					require.Fail(t, "timed out while setting up reconnection")
				}
				c.connStateCond.L.Lock()
				curConn := c.conn
				c.connStateCond.L.Unlock()
				if netconn.LocalAddr() == curConn.(*conn).socket.LocalAddr() {
					break
				}
			}
		}

		err := publishMessage(pubsterURL, m)
		require.NoError(t, err)

		msg, err := getMessage(c, 10*time.Second)
		require.NoError(t, err, "failed to retrieve message=%d", i)
		recvd = append(recvd, msg)
	}

	assert.Equal(t, n, len(recvd))
	assert.Equal(t, sent, recvd)

	assert.NoError(t, c.Close())
}

func TestSmoothReconnectIntegration(t *testing.T) {
	// Use a newConn func that lets us grab a reference to the
	// underlying *conn.
	var curConn *conn
	connCount := new(int64)
	newConn := func() (pubsubConn, error) {
		atomic.AddInt64(connCount, 1)
		dialer := &websocket.Dialer{TLSClientConfig: nil}
		socket, _, err := dialer.Dial(pubsubURL, nil)
		if err != nil {
			return nil, err
		}
		c := newPubsubConn(socket)
		curConn = c.(*conn)
		return c, nil
	}

	c := newClient(newConn)
	c.maxReconnectSleep = time.Nanosecond

	// This topic name is special: darklaunch allows topics that start
	// with 'pubsubtest.'
	const topic = "pubsubtest.topic3"
	assert.NoError(t, c.Subscribe(topic, ""))

	// In one goroutine, publish a message every 50ms for 4 seconds.
	//
	// In a second goroutine, retrieve pubsub messages from the client
	// store them. We eventually want every published message to appear
	// in our result set.
	//
	// After two seconds, spoof a "RECONNECT" message, getting the
	// client to reconnect. We should transfer to a new connection
	// smoothly without any dropped or duplicated messages.
	doneSending := make(chan struct{})
	doneRecving := make(chan struct{})
	var recvd []*Message
	go func() {
		defer close(doneRecving)
		for {
			msg, err := getMessage(c, time.Second)
			if err != nil {
				select {
				case <-doneSending:
					return
				default:
					t.Errorf("timed out waiting for message")
				}
			}
			recvd = append(recvd, msg)
		}
	}()

	var sent []*Message
	go func() {
		defer close(doneSending)
		publishTicker := time.NewTicker(50 * time.Millisecond)
		publishDone := time.NewTimer(4 * time.Second)
		i := 0
		for {
			i++
			select {
			case <-publishTicker.C:
				// Build a message
				msgBody := fmt.Sprintf("msg-%d", i)
				m := &Message{Topic: topic, Message: msgBody}

				// Record that we're publishing it for later
				sent = append(sent, m)

				// and actually do the publish
				err := publishMessage(pubsterURL, m)
				assert.NoError(t, err)
			case <-publishDone.C:
				publishTicker.Stop()
				return
			}
		}
	}()

	time.Sleep(2 * time.Second)
	assert.EqualValues(t, 1, atomic.LoadInt64(connCount))
	// Pretend we received a RECONNECT message
	curConn.recvReconnect()

	// Wait for send and recv loops to finish
	<-doneSending
	<-doneRecving

	assert.NoError(t, c.Close())
	assert.Equal(t, sent, recvd)
	assert.EqualValues(t, 2, atomic.LoadInt64(connCount))
}

func publishMessage(url string, m *Message) error {
	type publishMsg struct {
		Topics []string `json:"topics"`
		Data   string   `json:"data"`
	}
	pm := &publishMsg{
		Topics: []string{m.Topic},
		Data:   m.Message,
	}
	body, err := json.Marshal(pm)
	if err != nil {
		return err
	}
	_, err = http.Post(url, "application/json", bytes.NewBuffer(body))
	if err != nil {
		return err
	}
	return nil
}

// retrieve a message within a timeout. Fail the test if we don't get
// it in time.
func getMessage(c Client, timeout time.Duration) (*Message, error) {
	var (
		msg *Message
		err error
	)

	ch := make(chan struct{})
	go func() {
		msg, err = c.NextMessage()
		close(ch)
	}()
	select {
	case <-ch:
	case <-time.After(timeout):
		return nil, errors.New("timed out waiting for message")
	}
	return msg, err
}
