package pubsubclient

import (
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

var errReconnectRequested = errors.New("reconnect requested")

type pubsubConn interface {
	send(typedMessage) error
	// errors is a channel of errors received while handling the
	// connection. When the pubsubConn is closed, the errors channel
	// should be closed.
	errors() chan error

	// forwardTo sets up the channel that the pubsubConn should write its
	// messages into.
	forwardTo(chan<- *Message)

	close() error
}

func newPubsubConn(socket *websocket.Conn) pubsubConn {
	c := newConn(socket)
	go c.run()
	return c
}

// creates a new conn without running it, useful for tests
func newConn(socket *websocket.Conn) *conn {
	const buffer = 1000
	c := &conn{
		socket:          socket,
		errch:           make(chan error, 3), // 3 because run, sendLoop, recvLoop are the 3 writers to errch
		recvQueue:       make(chan []byte, buffer),
		sendQueue:       make(chan typedMessage, buffer),
		responses:       make(chan *responseMsg, buffer),
		responseTimeout: 10 * time.Second,
		messagesCh:      nil,
		pingInterval:    4 * time.Minute,
		pongTimeout:     30 * time.Second,
		pongs:           make(chan struct{}, 5),
		pongTimeouts:    make(chan struct{}, 1),
		closeSignal:     make(chan struct{}),
		closeTimeout:    5 * time.Second,
	}
	return c
}

// conn represents a connection to a pubsub edge. It manages the
// low-level details of the connection, like PINGing, PONGing, and
// waiting for RESPONSE messages to LISTEN and UNLISTEN requests.
//
// The conn should be managed by a parent connManager. The conn needs to be
// able to tell the connManager when it needs to reconnect, and the connManager needs to be able to pull
type conn struct {
	socket *websocket.Conn

	// messagesCh is the destination for parsed messages, set with the
	// forwardTo method.
	messagesLock sync.Mutex
	messagesCh   chan<- *Message

	// errch receives fatal errors that indicate the connection is
	// permanently broken and needs to be rebuilt.
	errch chan error

	// recvQueue and sendQueue hold the queued-up messages for the send
	// and receive loops.
	recvQueue chan []byte
	sendQueue chan typedMessage

	// responses holds the responses to LISTEN and UNLISTEN messages.
	responses       chan *responseMsg
	responseTimeout time.Duration

	// pingInterval is the frequence that the client should send PING
	// messages to the server.
	pingInterval time.Duration

	// pongTimeout is the time that the client should wait for a PONG
	// message after it sends a PING. If it doesn't receive one within
	// this time limit, it will mark an error.
	pongTimeout time.Duration

	// pongs and pongTimeouts are used to communicate internally when
	// the client is waiting for a PONG response to its PING.
	pongs        chan struct{}
	pongTimeouts chan struct{}

	// closeSignal will be closed when the connection is being torn down.
	closeSignal chan struct{}

	// closeTimeout is the length of time to wait for the remote end to
	// let us close the connection in a normal manner.
	closeTimeout time.Duration

	// wg keeps track of the send and recv loops.
	wg sync.WaitGroup
}

func (c *conn) errors() chan error { return c.errch }

func (c *conn) forwardTo(ch chan<- *Message) {
	c.messagesLock.Lock()
	defer c.messagesLock.Unlock()
	c.messagesCh = ch
}

func (c *conn) messages() chan<- *Message {
	c.messagesLock.Lock()
	defer c.messagesLock.Unlock()
	return c.messagesCh
}

// send sends a LISTEN or UNLISTEN message and waits for the
// RESPONSE from the server. It can return an error if its connection
// to the server breaks in any way, or if it times out waiting for a
// RESPONSE, or if the RESPONSE message indicates an error.
func (c *conn) send(m typedMessage) error {
	var nonce string
	switch v := m.(type) {
	case *listenMsg:
		nonce = v.Nonce
	case *unlistenMsg:
		nonce = v.Nonce
	default:
		return fmt.Errorf("invalid message type %T", m)
	}

	c.sendQueue <- m
	t := time.NewTimer(c.responseTimeout)
	defer t.Stop()

	for {
		select {
		case resp := <-c.responses:
			if resp.Nonce != nonce {
				// re-enqueue the response so some other listener can get it
				c.recvResponse(resp)
				continue
			}
			if resp.Error != "" {
				return errors.New(resp.Error)
			}
			return nil
		case <-t.C:
			return errors.New("response timed out")
		case <-c.closeSignal:
			return ErrClosed
		}
	}
}

// close shuts down the websocket connection. All future reads and
// sendMsgs will return errors. It can return an error if it does not
// shut down cleanly. It is an application error to call close()
// twice.
func (c *conn) close() error {
	err := c.closeSocket()
	close(c.closeSignal)
	c.wg.Wait()
	close(c.errch)
	return err
}

func (c *conn) closeSocket() error {
	err := c.socket.WriteControl(
		websocket.CloseMessage,
		websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""),
		time.Now().Add(c.closeTimeout),
	)
	if err != nil {
		if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
			return nil
		}
		_ = c.socket.Close()
		return err
	}
	return nil
}

func (c *conn) run() {
	c.wg.Add(2)
	go c.recvLoop()
	go c.sendLoop()
	pingTicker := time.NewTicker(c.pingInterval)
	for {
		select {
		case <-c.closeSignal:
			return
		case raw := <-c.recvQueue:
			err := c.handleRawMessage(raw)
			if err != nil {
				c.errch <- err
				return
			}
		case <-pingTicker.C:
			c.sendPing()
		case <-c.pongTimeouts:
			err := errors.New("timed out while waiting for pong")
			if err != nil {
				c.errch <- err
				return
			}
		}
	}
}

func (c *conn) recvLoop() {
	defer c.wg.Done()
	for {
		_, raw, err := c.socket.ReadMessage()
		if err != nil {
			if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
				return
			}
			c.errch <- err
			return
		}
		c.recvQueue <- raw
	}
}

func (c *conn) sendLoop() {
	defer c.wg.Done()
	for {
		var msg typedMessage
		select {
		case msg = <-c.sendQueue:
		case <-c.closeSignal:
			return
		}
		body, err := json.Marshal(msg)
		if err != nil {
			c.errch <- err
			return
		}
		err = c.socket.WriteMessage(websocket.TextMessage, body)
		if err != nil {
			if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
				return
			}
			c.errch <- err
			return
		}
	}
}

// handleRawMessage unmarshals a raw []byte message and then handles
// its contents. It can return an error if the message cannot be
// unmarshaled.
func (c *conn) handleRawMessage(raw []byte) error {
	msg, err := unmarshal(raw)
	if err != nil {
		return err
	}
	c.handleMessage(msg)
	return nil
}

// handleMessage is a simple routing method to execute the appropriate
// method for a message.
func (c *conn) handleMessage(msg typedMessage) {
	switch m := msg.(type) {
	case *pongMsg:
		c.recvPong()
	case *reconnectMsg:
		c.recvReconnect()
	case *messageMsg:
		c.recvMsg(&m.Data)
	case *responseMsg:
		c.recvResponse(m)
	}
}

func (c *conn) recvReconnect() {
	select {
	case c.errch <- errReconnectRequested:
	default:
		warnf("dropping RECONNECT message because c.errors is overflowing")
	}
}

// recvPong does a non-blocking send to c.pongs to signal that a pong
// was received.
func (c *conn) recvPong() {
	select {
	case c.pongs <- struct{}{}:
	default:
		warnf("dropping PONG message %+v because c.pongs is overflowing", pong)
	}
}

// recvMsg adds msg to the c.messages queue, if there is space
// available. If not, the message is dropped.
func (c *conn) recvMsg(msg *Message) {
	ch := c.messages()
	if ch == nil {
		warnf("dropping message %+v because c.messages() is nil", msg)
		return
	}
	select {
	case ch <- msg:
	default:
		warnf("dropping message %+v because c.messages is overflowing", msg)
	}
}

// recvResponse adds resp to the c.responses queue, if there is space
// available. If not, the response is dropped.
func (c *conn) recvResponse(resp *responseMsg) {
	select {
	case c.responses <- resp:
	default:
		warnf("dropping RESPONSE message %+v because c.responses is overflowing", resp)
	}
}

// sendPing sends a PING message and sets up a listener for a PONG
// response which will timeout after conn.pongTimeout.
func (c *conn) sendPing() {
	c.sendQueue <- &pingMsg{msg{ping}}

	go func() {
		// We're listening for a PONG. we'll be satisfied with a
		// stateChange too, though. If we don't get either, do a
		// non-blocking send on c.pongTimeouts.
		select {
		case <-time.After(c.pongTimeout):
			select {
			case c.pongTimeouts <- struct{}{}:
			default:
			}
		case <-c.pongs:
			return
		}
	}()
}
