package pubsubclient

import (
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"net/url"
	"sync"

	"github.com/gorilla/websocket"
)

type mockPubsubServer struct {
	upgrader *websocket.Upgrader

	sync.Mutex
	n        int
	disabled bool

	recv chan typedMessage
	send chan typedMessage

	forcedClose chan struct{}

	wg sync.WaitGroup
}

func newMockPubsubServer() *mockPubsubServer {
	return &mockPubsubServer{
		upgrader:    &websocket.Upgrader{},
		recv:        make(chan typedMessage, 1000),
		send:        make(chan typedMessage, 1),
		forcedClose: make(chan struct{}, 1),
	}
}

func runMockPubsubServer() (*mockPubsubServer, *httptest.Server) {
	s := newMockPubsubServer()
	ts := httptest.NewTLSServer(s)

	url, _ := url.Parse(ts.URL)
	url.Scheme = "wss"
	ts.URL = url.String()
	ts.TLS.InsecureSkipVerify = true

	return s, ts
}

func (s *mockPubsubServer) sendMsg(msg typedMessage) {
	s.send <- msg
}

func (s *mockPubsubServer) nConnections() int {
	s.Lock()
	defer s.Unlock()
	return s.n
}

func (s *mockPubsubServer) disable() {
	s.Lock()
	defer s.Unlock()
	s.disabled = true
}

func (s *mockPubsubServer) enable() {
	s.Lock()
	defer s.Unlock()
	s.disabled = false
}

// Immediately terminate any outstanding websocket connections
func (s *mockPubsubServer) forceClose() {
	select {
	case s.forcedClose <- struct{}{}:
	default:
	}
}

func (s *mockPubsubServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if s.disabled {
		return
	}
	s.wg.Add(1)
	defer s.wg.Done()

	conn, err := s.upgrader.Upgrade(w, r, nil)
	if err != nil {
		return
	}
	defer conn.Close()

	s.Lock()
	s.n++
	s.Unlock()

	var wg sync.WaitGroup
	done := make(chan struct{})

	// Have a goroutine which can just force the connection to die
	wg.Add(1)
	go func() {
		defer wg.Done()
		select {
		case <-done:
			return
		case <-s.forcedClose:
			conn.Close()
		}
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			_, p, err := conn.ReadMessage()
			if err != nil {
				if websocket.IsUnexpectedCloseError(err, websocket.CloseNormalClosure) {
					warnf("[mockPubsubServer] conn.ReadMessage() err: %s", err)
				}
				close(done)
				s.Lock()
				conn.WriteMessage(websocket.CloseMessage, []byte{})
				s.Unlock()
				break
			}
			msg, err := unmarshal(p)
			if err != nil {
				warnf("[mockPubsubServer] unmarshal err: %s", err)
				close(done)
				return
			}
			s.recv <- msg
		}
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case <-done:
				return
			case msg := <-s.send:
				bytes, err := json.Marshal(msg)
				if err != nil {
					warnf("[mockPubsubServer] json.Marshal err: %s", err)
					return
				}

				s.Lock()
				if err := conn.WriteMessage(websocket.TextMessage, bytes); err != nil {
					s.Unlock()
					warnf("[mockPubsubServer] conn.WriteMessage err: %s", err)
					return
				}
				s.Unlock()

			}
		}
	}()

	wg.Wait()
}
