package multiplex

import (
	"context"
	"net/http"
	"strconv"
	"strings"
	"time"

	telemetry "code.justin.tv/amzn/TwitchTelemetry"
	"code.justin.tv/live/autohost/internal/metrics"

	"code.justin.tv/creator-collab/log"
	"code.justin.tv/creator-collab/log/errors"
	"code.justin.tv/foundation/twitchclient"
	"code.justin.tv/video/multiplex/rpc/multiplex"
	"code.justin.tv/video/multiplex/rpc/multiplexv2"
	"code.justin.tv/video/usherapi/rpc/usher"

	"github.com/cactus/go-statsd-client/statsd"
)

const (
	// Maximum number that multiplex supports for calls to ChannelPropertiesList
	channelPropertiesListLimit = 100

	// Timeout for calls to Multiplex. These are set based on the deadlines defined at
	// https://git-aws.internal.justin.tv/video/multiplex/blob/master/internal/config/production-internal.yml
	channelPropertiesListTimeout = 10 * time.Second
)

// Client is a multiplex client
type Client interface {
	FilterLiveChannels(ctx context.Context, channelIDs []string) ([]string, error)
	LiveChannels(ctx context.Context) (map[string]bool, error)
}

type clientImpl struct {
	client         multiplex.Multiplex
	clientv2       multiplexv2.MultiplexV2
	logger         log.Logger
	sampleReporter *telemetry.SampleReporter
}

// NewClient returns a new multiplex client
func NewClient(url string, statsClient statsd.Statter, sampleReporter *telemetry.SampleReporter, logger log.Logger) Client {
	httpClient := twitchclient.NewHTTPClient(twitchclient.ClientConf{
		Host:           url,
		Stats:          statsClient,
		StatNamePrefix: "service.multiplex",
		TimingXactName: "multiplex_service",
		Transport: twitchclient.TransportConf{
			MaxIdleConnsPerHost: 20,
		},
		RoundTripperWrappers: []func(http.RoundTripper) http.RoundTripper{
			metrics.NewTwirpClientMiddlewareWrapper(&metrics.TwirpClientMiddlewareWrapperConfig{
				SampleReporter: sampleReporter,
			}),
		},
	})
	httpClient2 := twitchclient.NewHTTPClient(twitchclient.ClientConf{
		Host:           url,
		Stats:          statsClient,
		StatNamePrefix: "service.multiplex",
		TimingXactName: "multiplex_service",
		Transport: twitchclient.TransportConf{
			MaxIdleConnsPerHost: 10,
		},
		RoundTripperWrappers: []func(http.RoundTripper) http.RoundTripper{
			metrics.NewTwirpClientMiddlewareWrapper(&metrics.TwirpClientMiddlewareWrapperConfig{
				SampleReporter: sampleReporter,
			}),
		},
	})

	return &clientImpl{
		client:         multiplex.NewMultiplexProtobufClient(url, httpClient),
		clientv2:       multiplexv2.NewMultiplexV2ProtobufClient(url, httpClient2),
		logger:         logger,
		sampleReporter: sampleReporter,
	}
}

func (c *clientImpl) FilterLiveChannels(ctx context.Context, channelIDs []string) ([]string, error) {
	ctx = twitchclient.WithReqOpts(ctx, twitchclient.ReqOpts{
		StatName:       "channel_properties_list",
		StatSampleRate: 1,
	})
	ctx, cancelFunc := context.WithTimeout(ctx, channelPropertiesListTimeout)
	defer cancelFunc()

	res, err := c.client.ChannelPropertiesList(ctx, &usher.ChannelPropertiesListRequest{
		Limit:     channelPropertiesListLimit,
		ChannelId: strings.Join(channelIDs, ","),
	})
	if err != nil {
		return nil, errors.Wrap(err, "multiplex - filter live channels failed", errors.Fields{
			"channel_ids": channelIDs,
		})
	}
	liveChannels := make([]string, 0, len(res.GetChannelProperties()))
	for _, ch := range res.GetChannelProperties() {
		channelID := ch.GetChannelId()
		if channelID != 0 && ch.GetBroadcaster() != "spectre" {
			liveChannels = append(liveChannels, strconv.FormatInt(channelID, 10))
		}
	}

	// We need to dedupe the results because of a potential bug in usher that causes VP9 channels to be returned twice; see ING-4961
	return dedupeStrSlice(liveChannels), nil
}

// dedupeStrSlice removes duplicate elements from a string slice
func dedupeStrSlice(s []string) []string {
	m := make(map[string]bool, len(s))
	for _, item := range s {
		if _, ok := m[item]; !ok {
			m[item] = true
		}
	}
	result := make([]string, len(m))
	i := 0
	for k := range m {
		result[i] = k
		i++
	}
	return result
}

func (c *clientImpl) LiveChannels(ctx context.Context) (map[string]bool, error) {
	ctx = twitchclient.WithReqOpts(ctx, twitchclient.ReqOpts{
		StatName:       "channel_properties_live_hls_channels",
		StatSampleRate: 1,
	})

	res, err := c.clientv2.ChannelPropertiesLiveHLSChannels(ctx, &multiplexv2.ChannelPropertiesLiveHLSChannelsRequest{
		Request: &usher.ChannelPropertiesLiveHLSChannelsRequest{},
	})
	if err != nil {
		return nil, errors.Wrap(err, "multiplex - get live channels failed")
	}

	for _, regionRes := range res.GetResponses() {
		meta := regionRes.GetMeta()
		if meta.GetErrorOccurred() {
			return nil, errors.New("multiplex - get live channels failed", errors.Fields{
				"origin":        regionRes.GetOrigin(),
				"status_code":   meta.GetStatusCode(),
				"error_message": meta.GetErrorMessage(),
			})
		}
	}

	live := map[string]bool{}
	for _, regionRes := range res.GetResponses() {
		liveChannels := regionRes.GetResponse().GetChannels()
		for _, ch := range liveChannels {
			live[ch] = true
		}

		c.reportLiveChannelCountFromOrigin(regionRes.GetOrigin(), len(liveChannels))
	}
	liveCount := len(live)

	c.sampleReporter.Report("MultiplexTotalLiveChannelCount", float64(liveCount), telemetry.UnitCount)

	if liveCount == 0 {
		return nil, errors.New("multiplex - get live channels failed, got empty response")
	}

	return live, nil
}

func (c *clientImpl) reportLiveChannelCountFromOrigin(origin string, count int) {
	metricName := "MultiplexLiveChannelCount"

	if origin == "" {
		c.logger.Error(errors.New("building metric failed - no origin", errors.Fields{
			"metric_name": metricName,
		}))
		return
	}

	builder := c.sampleReporter.SampleBuilder // make a copy that we can modify

	if builder.Dimensions == nil {
		builder.Dimensions = make(telemetry.DimensionSet)
	}
	builder.Dimensions["Origin"] = origin

	sample, err := builder.Build(metricName, float64(count), telemetry.UnitCount)
	if err != nil {
		c.logger.Error(errors.Wrap(err, "building metric failed", errors.Fields{
			"metric_name": metricName,
		}))
		return
	}

	c.sampleReporter.ObserveSample(sample)
}
