package gqlsubs

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"time"

	"github.com/cep21/circuit"
	"github.com/cep21/circuit/closers/hystrix"
	"go.uber.org/zap"

	"code.justin.tv/websocket-edge/server/internal/environment"
	"code.justin.tv/websocket-edge/server/internal/logs"
	"code.justin.tv/websocket-edge/server/internal/metrics"
	"code.justin.tv/websocket-edge/server/internal/util"
	"code.justin.tv/websocket-edge/server/protocol"
)

const (
	metricError   = "GQLSubsMsgSendError"
	metricSuccess = "GQLSubsMsgSend"
	subsFmtStr    = "%s/message"
)

type Client interface {
	SendMessage(ctx context.Context, sessID string, clientAddr string, msgBody []byte) error
}

type client struct {
	circuit  *circuit.Circuit
	httpDoer util.HTTPDoer
	subsURL  string
	logger   logs.Logger
	statter  metrics.Statter
}

func New(logger logs.Logger, statter metrics.Statter, doer util.HTTPDoer, subsBaseURL string) *client {
	m := circuit.Manager{}
	circuitName := "GQLSubsMsgSend"
	circuitStatter := metrics.NewCircuitStatter(circuitName, statter)
	c := m.MustCreateCircuit(circuitName, circuit.Config{
		Execution: circuit.ExecutionConfig{
			Timeout:               1 * time.Second,
			MaxConcurrentRequests: 10000,
		},
		General: circuit.GeneralConfig{
			OpenToClosedFactory: hystrix.CloserFactory(hystrix.ConfigureCloser{
				SleepWindow:                  500 * time.Millisecond,
				HalfOpenAttempts:             20,
				RequiredConcurrentSuccessful: 5,
			}),
			ClosedToOpenFactory: hystrix.OpenerFactory(hystrix.ConfigureOpener{
				ErrorThresholdPercentage: 50,
				RequestVolumeThreshold:   20,
			}),
		},
		Metrics: circuit.MetricsCollectors{
			Run:      []circuit.RunMetrics{circuitStatter},
			Fallback: []circuit.FallbackMetrics{circuitStatter},
			Circuit:  []circuit.Metrics{circuitStatter},
		},
	})
	return &client{
		circuit:  c,
		httpDoer: doer,
		subsURL:  fmt.Sprintf(subsFmtStr, subsBaseURL),
		logger:   logger,
		statter:  statter,
	}
}

func (c *client) doRequest(ctx context.Context, req *http.Request) error {
	resp, err := c.httpDoer.Do(req)
	if err != nil {
		c.statter.IncrementErr(metricError, "ConnectionErr")
		c.logger.Error("error sending subs request", zap.Error(err))
		return err
	}
	defer func() {
		_, err := io.Copy(ioutil.Discard, resp.Body)
		if err != nil {
			c.statter.IncrementErr(metricError, "DiscardResponseBody")
		}
		err = resp.Body.Close()
		if err != nil {
			c.statter.IncrementErr(metricError, "CloseResponseBody")
		}
	}()

	if resp.StatusCode != http.StatusOK {
		c.statter.IncrementErr(metricError, fmt.Sprintf("HTTPErr%d", resp.StatusCode))
		c.logger.Error(fmt.Sprintf("Received bad status code of %d", resp.StatusCode))
		return err
	}

	c.statter.Increment(metricSuccess)
	return nil
}

func (c *client) SendMessage(ctx context.Context, sessID string, clientIP string, msgBody []byte) error {
	body := &protocol.ClientToServiceMessage{
		SessionID:   sessID,
		Body:        string(msgBody),
		ClientIP:    clientIP,
		HostAddress: environment.HostAddress(),
	}
	bodyBytes, err := json.Marshal(body)
	if err != nil {
		c.statter.IncrementErr(metricError, "MarshalErr")
		return err
	}
	req, err := http.NewRequest(http.MethodPost, c.subsURL, bytes.NewBuffer(bodyBytes))
	if err != nil {
		c.statter.IncrementErr(metricError, "BuildRequestErr")
		return err
	}

	// We ignore the error because everything except library errors has already been recorded.
	return c.circuit.Execute(ctx, func(ctx context.Context) error {
		return c.doRequest(ctx, req)
	}, nil)
}

// Noop gql subs client.
type Noop struct{}

func (*Noop) SendMessage(ctx context.Context, sessID string, clientAddr string, msgBody []byte) error {
	return nil
}
