package spade

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"expvar"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"code.justin.tv/hygienic/errors"
)

// Event represents a spade event
type Event struct {
	Name string `json:"event"`
	// Properties should be a JSON serializable struct
	Properties interface{} `json:"properties"`
}

func encodeBatchOfEvents(events ...Event) (string, error) {
	jb, err := json.Marshal(events)
	if err != nil {
		return "", err
	}

	jb64 := base64.URLEncoding.EncodeToString(jb)
	return jb64, nil
}

// HTTPRequestDoer can execute http requests
type HTTPRequestDoer interface {
	// Do should be like http.Client.Do
	Do(req *http.Request) (*http.Response, error)
}

var _ HTTPRequestDoer = &http.Client{}

// Logger outputs values
type Logger interface {
	Log(vals ...interface{})
}

// StatSender is deprecated. Use Reporter instead
type StatSender interface {
	Inc(string, int64, float32) error
}

// TaggingSubStatter is deprecated. Use Reporter instead
type TaggingSubStatter interface {
	IncD(key string, dimensions map[string]string, v int64)
	TimingDurationD(key string, dimensions map[string]string, v time.Duration)
}

// HTTPError represents an HTTP request error
type HTTPError struct {
	StatusCode int
	Body       string
}

func (h HTTPError) Error() string {
	return fmt.Sprintf("unexpected response status %d (%s)", h.StatusCode, h.Body)
}

var _ error = HTTPError{}

// Client sends data to spade
type Client struct {
	Config     Config
	HTTPClient HTTPRequestDoer `nilcheck:"nodepth"`
	Logger     Logger
	Reporter   Reporter
	// Deprecated: Use a Reporter created by an implementation in the spademetrics package
	Statter StatSender
	// Deprecated: Use a Reporter created by an implementation in the spademetrics package
	TaggingStats TaggingSubStatter

	Backlog chan Event

	closeSignal  chan struct{}
	doneDraining chan struct{}

	stats struct {
		TotalEventsSent int64
		AsyncErrors     int64
	}
}

// Vars allows client to expose expvar info
func (c *Client) Vars() expvar.Var {
	return expvar.Func(func() interface{} {
		return map[string]interface{}{
			"backlog_size":      len(c.Backlog),
			"total_events_sent": atomic.LoadInt64(&c.stats.TotalEventsSent),
			"async_errors":      atomic.LoadInt64(&c.stats.AsyncErrors),
		}
	})
}

// Setup should be called before you use the client
func (c *Client) Setup() error {
	c.doneDraining = make(chan struct{})
	c.closeSignal = make(chan struct{})
	if c.Backlog == nil {
		c.Backlog = make(chan Event, c.Config.GetBacklogSize())
	}
	if c.HTTPClient == nil {
		c.HTTPClient = http.DefaultClient
	}
	if c.Reporter == nil {
		c.Reporter = &noopReporter{}
	}
	return nil
}

func (c *Client) log(vals ...interface{}) {
	if c.Logger != nil {
		c.Logger.Log(vals...)
	}
}

// Deprecated: Use Reporter instead
func (c *Client) incStat(stat string, value int64, rate float32) {
	if c.Statter != nil {
		err := c.Statter.Inc(stat, value, rate)
		if err != nil {
			c.log("err", "failure sending stats", "staterr", err)
		}
	}
}

func (c *Client) emitSendSuccess(timeSince time.Duration) {
	c.Reporter.SendSuccess(timeSince)

	// Deprecated: Use Reporter.SendSuccess
	c.incStat("send_timing.success", 1, 1.0)
	if c.TaggingStats != nil {
		// Deprecated: Use Reporter.SendSuccess
		c.TaggingStats.TimingDurationD("send_timing", c.statTags("success"), timeSince)
	}
}

func (c *Client) emitSendFailure(timeSince time.Duration, err error) {
	c.Reporter.SendFailure(timeSince, err)

	// Deprecated: Use Reporter.SendFailure
	c.incStat("send_timing.fail", 1, 1.0)
	if c.TaggingStats != nil {
		// Deprecated: Use Reporter.SendFailure
		c.TaggingStats.TimingDurationD("send_timing", c.statTags("fail"), timeSince)
	}
}

func (c *Client) incRequestCount() {
	c.Reporter.RequestCount()

	// Deprecated: Use Reporter.RequestCount
	c.incStat("message.request_count", 1, 1.0)
	if c.TaggingStats != nil {
		// Deprecated: Use Reporter.RequestCount
		c.TaggingStats.IncD("request_count", c.statTags("message"), 1)
	}
}

// Deprecated: Use Reporter instead
func (c *Client) statTags(operationName string) map[string]string {
	return map[string]string{"origin": "spade", "Operation": operationName}
}

// Start should run in the background to drain the client's queue
func (c *Client) Start() error {
	defer func() {
		close(c.doneDraining)
	}()

	concurrency := c.Config.GetConcurrency()
	if concurrency <= 0 {
		concurrency = 1
	}
	// Any capacity value will work for the batches channel, though I've chosen to match Concurrency
	// here so that we can queue enough batches in the case Spade slows down, such that if all Spade
	// requests returned simultaneously -- the consumers would immediately send the next request.
	// I assume that we can read from the backlog and send to the batch channel faster than we can
	// send a Spade request, so matching Concurrency should be optimal to maximize throughput for
	// a specific Concurrency setting.
	batches := make(chan []Event, concurrency)
	var wg sync.WaitGroup
	c.startBatchConsumers(batches, concurrency, &wg)
	defer func() {
		close(batches)
		wg.Wait()
	}()

	return c.drainBacklog(batches)
}

func (c *Client) startBatchConsumers(batches <-chan []Event, concurrency int64, wg *sync.WaitGroup) {
	for i := int64(0); i < concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()

			for events := range batches {
				ctx := context.Background()
				ctx, cancel := context.WithTimeout(ctx, c.Config.GetTimeout())
				if err := c.BlockingSendEvents(ctx, events...); err != nil {
					// Note: This incStat can be removed because BlockingSendEvents reports on sent failures already
					c.incStat("errors.send_fail", int64(len(events)), 1.) // Deprecated: Use Reporter.SendFailure
					atomic.AddInt64(&c.stats.AsyncErrors, 1)
					c.log("err", err, "failed to drain spade queue")
				}
				cancel()
			}
		}()
	}
}

const sendDelay = 50 * time.Millisecond

func (c *Client) drainBacklog(batches chan<- []Event) error {
	for {
		eventsToSend := make([]Event, 1, c.Config.GetBatchSize())
		select {
		case eventsToSend[0] = <-c.Backlog:
			func() {
				timer := time.NewTimer(sendDelay)
				defer timer.Stop()
			outerLoop:
				for int64(len(eventsToSend)) < c.Config.GetBatchSize() {
					if c.Config.CombineSmallBatchesEnabled() {
						// We want to avoid sending Spade requests too quickly in the case we're not filling
						// up the Backlog quickly. Without this timeout, a high Concurrency setting would cause
						// many single Event requests to be sent to Spade.
						//
						// The timeout is only used when a batch is not yet full, so in case the Backlog is
						// filled up quickly it will send batches as fast as possible.
						select {
						case anotherEvent := <-c.Backlog:
							eventsToSend = append(eventsToSend, anotherEvent)
						case <-timer.C:
							break outerLoop
						}
					} else {
						select {
						case anotherEvent := <-c.Backlog:
							eventsToSend = append(eventsToSend, anotherEvent)
						default:
							break outerLoop
						}
					}
				}
				// Either the batch is full and timer is still running or the timer has expired.
				batches <- eventsToSend
			}()
		case <-c.closeSignal:
			return nil
		}
	}
}

// Close ends the client
func (c *Client) Close() error {
	close(c.closeSignal)
	<-c.doneDraining
	return nil
}

// VerifyEndpoint checks that spade works
func (c *Client) VerifyEndpoint(ctx context.Context) error {
	req, err := http.NewRequest("GET", c.Config.GetSpadeHost()+"/xarth", nil)
	if err != nil {
		return err
	}
	req = req.WithContext(ctx)
	resp, err := c.HTTPClient.Do(req)
	if err != nil {
		return err
	}
	defer func() {
		if e := resp.Body.Close(); e != nil {
			c.log("err", e, "failed to close response body")
		}
	}()
	if resp.StatusCode != http.StatusOK {
		return errors.Errorf("invalid status code %d", resp.StatusCode)
	}
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return err
	}
	if string(body) != "XARTH" {
		return errors.Errorf("invalid body %s", string(body))
	}
	return nil
}

// QueueEvents adds events to the internal queue.  Returns instantly if the queue is full.
func (c *Client) QueueEvents(events ...Event) {
	for _, event := range events {
		select {
		case c.Backlog <- event:
		default:
			c.Reporter.BacklogFull()
			c.incStat("errors.backlog_full", 1, 1.) // Deprecated: Use Reporter.BacklogFull
			atomic.AddInt64(&c.stats.AsyncErrors, 1)
			c.log("spade backlog full and cannot queue more events")
		}
	}
}

// BlockingSendEvents sends inline, not using the queue
func (c *Client) BlockingSendEvents(ctx context.Context, events ...Event) error {
	start := time.Now()
	err := c.doBlockingSendEvents(ctx, events...)
	dur := time.Since(start)
	if err != nil {
		c.emitSendFailure(dur, err)
	} else {
		c.emitSendSuccess(dur)
	}

	return err
}

func (c *Client) doBlockingSendEvents(ctx context.Context, events ...Event) error {
	encodedEvents, err := encodeBatchOfEvents(events...)
	if err != nil {
		return errors.Wrap(err, "cannot encode spade events")
	}
	vals := url.Values{}
	vals.Add("data", encodedEvents)
	req, err := http.NewRequest("POST", c.Config.GetSpadeHost()+"/track", strings.NewReader(vals.Encode()))
	if err != nil {
		return errors.Wrap(err, "cannot make request")
	}
	req.Header.Add("User-Agent", "hygienic-spade 1.0")
	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	req = req.WithContext(ctx)
	resp, err := c.HTTPClient.Do(req)
	c.incRequestCount()
	if err != nil {
		return errors.Wrap(err, "cannot execute spade HTTP request")
	}
	defer func() {
		if e := resp.Body.Close(); e != nil {
			c.log("err", e, "failed to close response body")
		}
	}()

	var bodyBuff bytes.Buffer
	if _, err := io.CopyN(&bodyBuff, resp.Body, 1024); err != nil && err != io.EOF {
		return errors.Wrap(err, "cannot read response body")
	}
	if _, err := io.Copy(ioutil.Discard, resp.Body); err != nil && err != io.EOF {
		return errors.Wrap(err, "cannot drain response body")
	}
	if resp.StatusCode != http.StatusNoContent {
		return HTTPError{StatusCode: resp.StatusCode, Body: bodyBuff.String()}
	}

	c.Reporter.Events(len(events))
	c.incStat("events", int64(len(events)), 0.1) // Deprecated: use Reporter.Events
	atomic.AddInt64(&c.stats.TotalEventsSent, int64(len(events)))
	return nil
}
