package metrics

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"sync"

	telemetry "code.justin.tv/amzn/TwitchTelemetry"
	metricsmiddleware "code.justin.tv/amzn/TwitchTelemetryMetricsMiddleware"
	"code.justin.tv/video/metrics-middleware/v2/operation"
)

var twitchClientOperationCtxKey = new(int)

// TwitchClientMiddleware is a [RoundTripper](https://golang.org/pkg/net/http/#RoundTripper) that is meant to
// instrument TwitchClient styled clients. It supports sending standard dependency availability and latency
// metrics to CloudWatch using the Twitch Telemetry libraries.
//
//     Metric names: DependencyDuration, DependencySuccess, DependencyClientError, DependencyServerError
//     Dimensions:
//       * Dependency (The client's service and operation name: e.g. twitchclient/UsersService:GetUser)
//       * Operation (The overall API operation being processed e.g. HostTarget)
//       * etc
//
// TwitchClients are usually created from a twitchclient.ClientConf struct.
// To add TwitchClientMiddleware, add a function to the ClientConf's RoundTripWrappers that wraps
// an underlying RoundTripper with an instance of TwitchClientMiddleware.
//
//     func NewClient(url string, sampleReporter *telemetry.SampleReporter) (Client, error) {
//         wrapRoundTripper := func(rt http.RoundTripper) http.RoundTripper {
//             return metrics.NewTwitchClientMiddleware(&metrics.TwitchClientMiddlewareConfig{
//                 SampleReporter: sampleReporter,
//                 Inner:          rt,
//             })
//         }
//	       clientConf := twitchhttp.ClientConf{
//	           Host: url,
//             RoundTripperWrappers: []func(http.RoundTripper) http.RoundTripper{
//			       wrapRoundTripper,
//		       },
//          }
//          // snip
//	    }
//
// The TwitchClient's methods will need to pass in its service and operation name to TwitchClientMiddleware.
// It can do so by putting it on the context using WithTwitchClientOperation.
//
//     func (c *clientImpl) GetUser(ctx context.Context, channelID, targetID string) error {
//         ctx = metrics.WithTwitchClientOperation(ctx, "UsersService", "GetUser")
//
//         // snip - create and send http request
//     }
type TwitchClientMiddleware struct {
	Inner   http.RoundTripper
	Starter *operation.Starter
}

type TwitchClientMiddlewareConfig struct {
	SampleReporter *telemetry.SampleReporter
	Inner          http.RoundTripper

	// Whether telemetry should send metrics after every request.
	// This should usually be set to false so that telemetry can buffer metrics.
	AutoFlush bool
}

func NewTwitchClientMiddleware(config *TwitchClientMiddlewareConfig) *TwitchClientMiddleware {
	opMonitor := &metricsmiddleware.OperationMonitor{
		SampleReporter: *config.SampleReporter,
		AutoFlush:      config.AutoFlush,
		MonitorConfig: metricsmiddleware.MonitorConfig{
			HTTPErrorCodes: true,
		},
	}
	opStarter := &operation.Starter{
		OpMonitors: []operation.OpMonitor{
			opMonitor,
		},
	}

	return &TwitchClientMiddleware{
		Inner:   config.Inner,
		Starter: opStarter,
	}
}

// TwitchClientMiddleware implements the http.RoundTripper interface.
var _ http.RoundTripper = &TwitchClientMiddleware{}

// RoundTrip executes a single HTTP transaction wrapped in a client operation.
func (client *TwitchClientMiddleware) RoundTrip(request *http.Request) (*http.Response, error) {
	ctx := request.Context()

	// Start an telemetry operation.
	twitchClientOp := twitchClientOperationFromContext(ctx)
	ctx, op := client.Starter.StartOp(ctx, operation.Name{
		Kind:   operation.KindClient,
		Group:  fmt.Sprintf("twitchclient/%s", twitchClientOp.serviceName),
		Method: twitchClientOp.methodName,
	})

	// Send the HTTP request.
	response, err := client.Inner.RoundTrip(request.WithContext(ctx))

	// End the telemetry operation.
	ctxErr := ctx.Err()
	switch {
	// If sending the request timed out, or was cancelled, there is no response to read.
	// End our operation with an appropriate status.
	case ctxErr == context.Canceled:
		op.SetStatus(operation.Status{Code: grpcCancelled})
		op.End()
	case ctxErr == context.DeadlineExceeded:
		op.SetStatus(operation.Status{Code: grpcDeadlineExceeded})
		op.End()

	// If the HTTP request could not be sent, there is no response to read.
	// End our operation with an internal error code.
	case err != nil:
		op.SetStatus(operation.Status{
			Code:    grpcInternal,
			Message: fmt.Sprintf("failed to do request: %v", err),
		})
		op.End()

	// If we were able to send the request and get a response, we want to end the operation after
	// the client finishes reading the body. To do that, we wrap the body's ReadCloser with a
	// custom ReadCloser that ends the op after the client finishes reading or closing the response.
	default:
		response.Body = newBodyMiddleware(response.Body, op, response.StatusCode)
	}

	return response, err
}

// bodyMiddleware wraps an HTTP response's body with a custom ReadCloser that ends a telemetry operation
// when the body has been read.
type bodyMiddleware struct {
	statusCode int
	op         *operation.Op
	inner      io.ReadCloser

	mutex               sync.Mutex
	endOperationCalled  bool
	lastUnexpectedError error
}

func newBodyMiddleware(inner io.ReadCloser, op *operation.Op, responseStatusCode int) *bodyMiddleware {
	return &bodyMiddleware{
		statusCode: responseStatusCode,
		op:         op,
		inner:      inner,
	}
}

// bodyMiddleware implements the io.ReadCloser interface.
var _ io.ReadCloser = &bodyMiddleware{}

func (m *bodyMiddleware) Read(p []byte) (int, error) {
	var n int
	var err error
	if m.inner != nil {
		n, err = m.inner.Read(p)
	}

	m.mutex.Lock()
	defer m.mutex.Unlock()

	// Record the last unexpected error.
	if err != io.EOF {
		m.lastUnexpectedError = err
	}

	// If Read returns io.EOF, we've read the entire stream. End the telemetry operation.
	if err == io.EOF {
		m.endOperation()
	}

	return n, err
}

func (m *bodyMiddleware) Close() error {
	var err error
	if m.inner != nil {
		err = m.inner.Close()
	}

	m.mutex.Lock()
	defer m.mutex.Unlock()

	m.endOperation()

	return err
}

func (m *bodyMiddleware) endOperation() {
	if m.endOperationCalled {
		return
	}
	m.endOperationCalled = true

	status := operation.Status{
		Code: httpStatusToGrpcCode(m.statusCode),
	}
	if m.lastUnexpectedError != nil {
		status = operation.Status{
			Code:    grpcInternal,
			Message: "failed to read response body: " + m.lastUnexpectedError.Error(),
		}
	}

	m.op.SetStatus(status)
	m.op.End()
}

type WrapperFunc func(http.RoundTripper) http.RoundTripper

type TwitchClientMiddlewareWrapperConfig struct {
	SampleReporter *telemetry.SampleReporter

	// Whether telemetry should send metrics after every request.
	// This should usually be set to false so that telemetry can buffer metrics.
	AutoFlush bool
}

// NewTwitchClientMiddlewareWrapper returns a function that takes a round tripper, and wraps it
// in TwitchClientMiddleware.
func NewTwitchClientMiddlewareWrapper(conf *TwitchClientMiddlewareWrapperConfig) WrapperFunc {
	return func(rt http.RoundTripper) http.RoundTripper {
		return NewTwitchClientMiddleware(&TwitchClientMiddlewareConfig{
			SampleReporter: conf.SampleReporter,
			Inner:          rt,
			AutoFlush:      conf.AutoFlush,
		})
	}
}

type twitchClientOperation struct {
	serviceName string
	methodName  string
}

// WithTwitchClientOperation attaches a client's service and method name to the context so that they
// can be accessed by TwitchClientMiddleware. These values are used to populate the Dependency dimension
// on CloudWatch metrics.
func WithTwitchClientOperation(ctx context.Context, serviceName, methodName string) context.Context {
	return context.WithValue(ctx, twitchClientOperationCtxKey, &twitchClientOperation{
		serviceName: serviceName,
		methodName:  methodName,
	})
}

func twitchClientOperationFromContext(ctx context.Context) *twitchClientOperation {
	val := ctx.Value(twitchClientOperationCtxKey)
	if val == nil {
		return &twitchClientOperation{
			serviceName: "Unknown",
			methodName:  "Unknown",
		}
	}

	return val.(*twitchClientOperation)
}
