package usher

import (
	"bytes"
	"context"
	"encoding/json"
	fmt "fmt"
	"net/http"
	url "net/url"
	"strconv"
	"strings"
	"time"

	"github.com/golang/protobuf/jsonpb"
	"github.com/golang/protobuf/ptypes"
	structpb "github.com/golang/protobuf/ptypes/struct"
	"github.com/golang/protobuf/ptypes/timestamp"
	"github.com/golang/protobuf/ptypes/wrappers"
	"github.com/pkg/errors"

	"code.justin.tv/common/alvin/restclient"
)

// NewUsherClient constructs a new usher client using Alvin for direct use with
// usher (as opposed to NewUsherJsonClient or NewUsherProtobufClient). The
// returned client includes support for the singlereply option.
//
// The singlereply method option is used to indiciate that an usher's response
// will be exactly one JSON Object wrapped in a JSON Array. When singlereply is
// set, this usher client will have the following behavior:
//
// - If Usher responds with an array of length == 1, the single element in that
//   array will be deserialized into the RPC's return message type
// - If Usher responds with an array of length != 1, an error will be returned
// - Any other response will also result in an error
//
// usherURL is the URL to the Usher API e.g. http://video-api.internal.justin.tv
// client is the HTTP client to use when communcating with Usher
func NewUsherClient(usherURL string, client *http.Client) (Usher, error) {
	usherURL, err := formatUsherURL(usherURL)
	if err != nil {
		return nil, fmt.Errorf("bad usherURL=\"%s\": %s", usherURL, err)
	}

	cl := new(http.Client)
	if client == nil {
		client = http.DefaultClient
	}

	// shallow copy
	*cl = *client

	rt, err := newUsherRoundTripper(cl.Transport)
	if err != nil {
		return nil, err
	}

	rt, err = restclient.RoundTripperFromProtos(rt,
		[]string{
			"code.justin.tv/video/usherapi/rpc/usher/usher.proto",
		})
	if err != nil {
		return nil, err
	}

	cl.Transport = rt

	c := NewUsherProtobufClient(usherURL, cl)

	return &alvinUsherClient{Usher: c}, nil

}

// formatUsherURL makes sure the given URL is valid and doesn't have a trailing slash
func formatUsherURL(usherURL string) (string, error) {
	url, err := url.Parse(usherURL)
	if err != nil {
		return "", fmt.Errorf("unable to parse given Usher URL: %s", err)
	}
	usherURL = url.String()
	// due to a quirk of the generated Twirp client, the given URL cannot have a trailing '/'
	usherURL = strings.TrimRight(usherURL, "/")
	return usherURL, nil
}

// alvinUsherClient is a wrapper around an Usher client which reshapes response
// objects to fully deserialize them.
type alvinUsherClient struct {
	Usher
}

func (a *alvinUsherClient) HLSTranscodeShowChannelAll(ctx context.Context, req *HLSTranscodeShowChannelRequest) (*ListOf_HLSTranscode, error) {
	resp, err := a.Usher.HLSTranscodeShowChannelAll(ctx, req)
	if err != nil {
		return resp, err
	}

	if err = unmarshalListOfHLSTranscode(resp); err != nil {
		return nil, err
	}
	return resp, nil
}

func (a *alvinUsherClient) HLSTranscodeShowChannel(ctx context.Context, req *HLSTranscodeShowChannelRequest) (*HLSTranscode, error) {
	resp, err := a.Usher.HLSTranscodeShowChannel(ctx, req)
	if err != nil {
		return resp, err
	}

	if err = unmarshalHLSTranscode(resp); err != nil {
		return nil, err
	}
	return resp, nil
}

func (a *alvinUsherClient) HLSTranscodeShow(ctx context.Context, req *HLSTranscodeShowRequest) (*HLSTranscode, error) {
	resp, err := a.Usher.HLSTranscodeShow(ctx, req)
	if err != nil {
		return resp, err
	}

	if err = unmarshalHLSTranscode(resp); err != nil {
		return nil, err
	}
	return resp, nil
}

func (a *alvinUsherClient) HLSTranscodeUpdate(ctx context.Context, req *HLSTranscodeUpdateRequest) (*HLSTranscode, error) {
	if req.Update != nil {
		err := marshalHLSTranscodeUpdate(req.Update)
		if err != nil {
			return nil, err
		}
	}

	resp, err := a.Usher.HLSTranscodeUpdate(ctx, req)
	if err != nil {
		return nil, err
	}

	if err = unmarshalHLSTranscode(resp); err != nil {
		return nil, err
	}
	return resp, nil
}

func (a *alvinUsherClient) HLSTranscodeKill(ctx context.Context, req *HLSTranscodeKillRequest) (*HLSTranscode, error) {
	resp, err := a.Usher.HLSTranscodeKill(ctx, req)
	if err != nil {
		return resp, err
	}

	if err = unmarshalHLSTranscode(resp); err != nil {
		return nil, err
	}
	return resp, nil
}

func (a *alvinUsherClient) NodeList(ctx context.Context, req *NodeListRequest) (*ListOf_Node, error) {
	resp, err := a.Usher.NodeList(ctx, req)
	if err != nil {
		return resp, err
	}

	if err = unmarshalNode(resp.Nodes); err != nil {
		return nil, err
	}
	return resp, nil
}

func (a *alvinUsherClient) NodeListEdge(ctx context.Context, req *NodeListEdgeRequest) (*ListOf_Node, error) {
	resp, err := a.Usher.NodeListEdge(ctx, req)
	if err != nil {
		return resp, err
	}

	if err = unmarshalNode(resp.Nodes); err != nil {
		return nil, err
	}
	return resp, nil
}

func (a *alvinUsherClient) ClusterList(ctx context.Context, req *ClusterListRequest) (*ListOf_Cluster, error) {
	resp, err := a.Usher.ClusterList(ctx, req)
	if err != nil {
		return resp, err
	}

	if err = unmarshalClusterTimestamps(resp.Clusters); err != nil {
		return nil, err
	}
	return resp, nil
}

func (a *alvinUsherClient) ChannelPropertiesList(ctx context.Context, req *ChannelPropertiesListRequest) (*ListOf_ChannelPropertiesListResponse, error) {
	resp, err := a.Usher.ChannelPropertiesList(ctx, req)
	if err != nil {
		return resp, err
	}
	if err = unmarshalChannelPropertiesListTimestamps(resp.ChannelProperties); err != nil {
		return nil, err
	}
	return resp, nil
}

func marshalHLSTranscodeUpdate(u *HLSTranscodeUpdate) error {
	if u.LvsMeta != nil {
		buf, err := marshalLVSMetadata(u.LvsMeta)
		if err != nil {
			return err
		}
		u.LvsMetadataRawString = &wrappers.StringValue{Value: string(buf)}
		u.LvsMeta = nil
	}

	if len(u.Renditions) > 0 {
		buf, err := marshalRenditionMetadata(u.Renditions)
		if err != nil {
			return err
		}
		u.RenditionMetaRawString = &wrappers.StringValue{Value: string(buf)}
		u.Renditions = nil
	}

	if u.EncryptionKeyIds != nil {
		buf, err := json.Marshal(u.EncryptionKeyIds)
		if err != nil {
			return err
		}
		u.EncryptionKeyIds = nil
		u.EncryptionKeyIdsRawString = &wrappers.StringValue{Value: string(buf)}
	}

	return nil
}

func marshalLVSMetadata(m *LVSMetadata) ([]byte, error) {
	var buf bytes.Buffer
	if err := (&jsonpb.Marshaler{}).Marshal(&buf, m); err != nil {
		return nil, errors.Wrap(err, "unable to marshal lvs_metadata")
	}

	return buf.Bytes(), nil
}

func marshalRenditionMetadata(renditions []*RenditionMetadata) ([]byte, error) {
	var renditionList []json.RawMessage
	for _, r := range renditions {
		var buf bytes.Buffer
		if err := (&jsonpb.Marshaler{}).Marshal(&buf, r); err != nil {
			return nil, errors.Wrap(err, "unable to marshal rendition_meta item")
		}
		renditionList = append(renditionList, json.RawMessage(buf.Bytes()))
	}
	out, err := json.Marshal(renditionList)
	if err != nil {
		return nil, errors.Wrap(err, "unable to marshal rendition_meta array")
	}
	return out, err
}

func unmarshalListOfHLSTranscode(list *ListOf_HLSTranscode) error {
	for _, transcode := range list.Transcodes {
		err := unmarshalHLSTranscode(transcode)
		if err != nil {
			return err
		}
	}
	return nil
}

// unmarshalHLSTranscode unmarshals nested JSON payloads within t.
func unmarshalHLSTranscode(t *HLSTranscode) error {
	if len(t.LvsMetadataRawString) > 0 {
		t.LvsMeta = new(LVSMetadata)
		err := jsonpb.UnmarshalString(t.LvsMetadataRawString, t.LvsMeta)
		if err != nil {
			return errors.Wrap(err, "unable to unmarshal lvs_metadata")
		}
		t.LvsMetadataRawString = ""
	}

	if len(t.RenditionMetaRawString) > 0 {
		var rawRenditions []json.RawMessage
		if err := json.Unmarshal([]byte(t.RenditionMetaRawString), &rawRenditions); err != nil {
			return errors.Wrap(err, "unable to unmarshal rendition_meta array")
		}
		for _, raw := range rawRenditions {
			var r RenditionMetadata
			if err := jsonpb.UnmarshalString(string(raw), &r); err != nil {
				return errors.Wrap(err, "unable to unmarshal rendition_meta item")
			}
			t.Renditions = append(t.Renditions, &r)
		}

		t.RenditionMetaRawString = ""
	}

	if len(t.EncryptionKeyIdsRawString) > 0 {
		var keys []string
		if err := json.Unmarshal([]byte(t.EncryptionKeyIdsRawString), &keys); err != nil {
			return errors.Wrap(err, "unable to unmarshal encryption_key_ids")
		}
		t.EncryptionKeyIds = keys
		t.EncryptionKeyIdsRawString = ""
	}

	return nil
}

// parseEpochSeconds attempts to turn the passed in json value into a protobuf
// timestamp. It a string serialized integer as well as a json number value.
func parseEpochSeconds(v *structpb.Value) (*timestamp.Timestamp, error) {
	var seconds int64

	if v == nil {
		return ptypes.TimestampProto(time.Unix(seconds, 0))
	}

	switch v := v.Kind.(type) {
	case *structpb.Value_NumberValue:
		seconds = int64(v.NumberValue)
	case *structpb.Value_StringValue:
		s, err := strconv.ParseInt(v.StringValue, 10, 64)
		if err != nil {
			return nil, errors.Wrapf(err, "invalid epoch time string %q", v.StringValue)
		}
		seconds = s
	default:
		return nil, errors.Errorf("invalid time value %v", v)
	}

	return ptypes.TimestampProto(time.Unix(seconds, 0))
}

// unmarshalNode parses and converts node fields to types that are easier
// for clients to use. The nodes returned from /node/list contain RFC3339
// timestamps and are already unmarshaled to the UpdateTime and CreateTime fields.
// The edge nodes returned from /node/list_edges return
// timestamps in epoch seconds (either as numbers or strings), which
// are parsed and stored in the same RFC3339 fields.
//
// Note that the /node/list endpoint returns ANSIC strings in addition to the RFC3339 timestamps
// and are not used due to their ambiguity. Thus, this parsing only supports the RFC3339 timestamps
// and epoch seconds (ANSIC timestamps would cause an error during epoch parsing).
func unmarshalNode(nodes []*Node) error {
	for _, n := range nodes {
		if n.UpdateTime == nil {
			ts, err := parseEpochSeconds(n.UpdateTimeAnsicOrSeconds)
			if err != nil {
				return errors.Wrap(err, "unable to parse update_time from node entry")
			}
			n.UpdateTime = ts
		}
		n.UpdateTimeAnsicOrSeconds = nil

		if n.CreateTime == nil {
			ts, err := parseEpochSeconds(n.CreateTimeAnsicOrSeconds)
			if err != nil {
				return errors.Wrap(err, "unable to parse create_time from node entry")
			}
			n.CreateTime = ts
		}

		n.CreateTimeAnsicOrSeconds = nil

		if n.ExperimentMetricsRaw != "" {
			var em map[string]float64
			if err := json.Unmarshal([]byte(n.ExperimentMetricsRaw), &em); err != nil {
				return errors.Wrap(err, "unable to parse experiment_metrics field")
			}
			n.ExperimentMetrics = em
			n.ExperimentMetricsRaw = ""
		}
	}
	return nil
}

// unmarshalClusterTimestamps handles properly converting cluster timestamp
// values when necesary
func unmarshalClusterTimestamps(clusters []*Cluster) error {
	for _, c := range clusters {
		ts, err := ptypes.TimestampProto(time.Unix(c.LinkInfoUpdateTimeSeconds, 0))
		if err != nil {
			return errors.Wrapf(err, "unexpected error making google.protobuf.Timestamp value from (%d)", c.LinkInfoUpdateTimeSeconds)
		}
		c.LinkInfoUpdateTime = ts
		c.LinkInfoUpdateTimeSeconds = 0
	}

	return nil
}

func unmarshalChannelPropertiesListTimestamps(channelProperties []*ChannelPropertiesListItem) error {
	for _, item := range channelProperties {
		if item.UpTime != "" {
			upTimeTime, err := time.Parse(time.RFC3339, item.UpTime)
			if err != nil {
				return errors.Wrapf(err, "could not parse time.Time from %s", item.UpTime)
			}
			upTimePtype, err := ptypes.TimestampProto(upTimeTime)
			if err != nil {
				return errors.Wrapf(err, "could not create ptypes.Timestamp from %s", item.UpTime)
			}
			item.UpTimeTimestamp = upTimePtype
		}
		if item.UpdatedOn != "" {
			updatedOnTime, err := time.Parse(time.RFC3339, item.UpdatedOn)
			if err != nil {
				return errors.Wrapf(err, "could not parse time.Time from %s", item.UpdatedOn)
			}
			updatedOnPtype, err := ptypes.TimestampProto(updatedOnTime)
			if err != nil {
				return errors.Wrapf(err, "could not create ptypes.Timestamp from %s", item.UpdatedOn)
			}
			item.UpdatedOnTimestamp = updatedOnPtype
		}
	}
	return nil
}
