package scanproto

import (
	"errors"
	"math/rand"
	"sync"
	"time"

	"code.justin.tv/release/trace/api"
	"code.justin.tv/release/trace/internal/awsutils"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/kinesis"
	"github.com/golang/protobuf/proto"
	"golang.org/x/net/context"
)

const (
	// Kinesis's DescribeStream API is restricted to describing at most
	// 10,000 shards per response.
	maxShardsPerDescribe = 10000
	// Maximum number of Kinesis DescribeStreams pages to read through
	maxDescribeStreamPages = 10
)

// implements scanproto.TransactionSource. Doesn't use the Multilang Daemon,
// so it shares no state with other processes.
type simpleKinesisTxSource struct {
	client *kinesis.Kinesis
	stream string

	txs    chan *api.Transaction
	errors chan error

	ctx       context.Context
	cancel    func()
	wg        sync.WaitGroup
	closeOnce sync.Once
}

func NewKinesisTransactionSource(ctx context.Context, client *kinesis.Kinesis, stream string) TransactionSource {
	ctx, cancel := context.WithCancel(ctx)

	cfg := awsutils.WithChitin(ctx, client.Config.Copy())

	client = kinesis.New(session.New(cfg))

	return &simpleKinesisTxSource{
		client: client,
		stream: stream,
		txs:    make(chan *api.Transaction, readBuffer),
		errors: make(chan error, 1),
		ctx:    ctx,
		cancel: cancel,
	}
}

func listShards(k *kinesis.Kinesis, stream string) (ids []string, err error) {
	req := &kinesis.DescribeStreamInput{
		StreamName: aws.String(stream),
		Limit:      aws.Int64(maxShardsPerDescribe),
	}

	resp, err := k.DescribeStream(req)
	if err != nil {
		return nil, err
	}
	if resp.StreamDescription == nil {
		return nil, errors.New("empty DescribeStream response")
	}
	ids = make([]string, len(resp.StreamDescription.Shards))
	for i, shard := range resp.StreamDescription.Shards {
		ids[i] = aws.StringValue(shard.ShardId)
	}

	// If the response indicates pagination is necessary, read
	// through the pages
	for i := 0; i < maxDescribeStreamPages && aws.BoolValue(resp.StreamDescription.HasMoreShards); i++ {
		req.ExclusiveStartShardId = aws.String(ids[len(ids)-1])
		resp, err = k.DescribeStream(req)
		if err != nil {
			return nil, err
		}
		if resp.StreamDescription == nil {
			return nil, errors.New("empty DescribeStream response")
		}

		for _, shard := range resp.StreamDescription.Shards {
			ids = append(ids, aws.StringValue(shard.ShardId))
		}
	}

	return ids, nil
}

func (s *simpleKinesisTxSource) Run() {
	ids, err := listShards(s.client, s.stream)
	if err != nil {
		s.errors <- err
		return
	}

	for _, id := range ids {
		s.wg.Add(1)
		sc := &shardConsumer{
			source:  s,
			ShardId: id,
		}
		go sc.consume()
	}

	s.wg.Wait()
}

func (s *simpleKinesisTxSource) reportErr(err error) {
	if s.ctx.Err() == nil {
		// We're shutting down. The user probably doesn't care about these
		// errors, since they're probably caused by the shutdown.
		s.errors <- err
	}
}

type shardConsumer struct {
	source *simpleKinesisTxSource

	ShardId        string
	Iterator       string
	SequenceNumber string
}

func (sc *shardConsumer) throughputExceededRetry(fn func() error) error {
	next := time.NewTimer(0)
	defer next.Stop()
	for {
		<-next.C
		resetTo(next, 1*time.Second-time.Duration(rand.Int63n(int64(500*time.Millisecond))))

		err := fn()
		if err == nil {
			return nil
		}
		if ae, ok := err.(awserr.Error); ok && ae.Code() == "ProvisionedThroughputExceededException" {
			// Back off, with jitter. Sufficient backoff is in our best
			// interest, since AWS seems to disable HTTP Keep-Alive when we
			// make requests too quickly.
			time.Sleep(1*time.Second - time.Duration(rand.Int63n(int64(500*time.Millisecond))))
			continue
		}
		if err != nil {
			sc.source.reportErr(err)
			return err
		}
	}
}

func (sc *shardConsumer) GetShardIterator() error {
	var (
		err      error
		itOutput *kinesis.GetShardIteratorOutput
	)

	itInput := &kinesis.GetShardIteratorInput{
		ShardId:    aws.String(sc.ShardId),
		StreamName: aws.String(sc.source.stream),
	}

	if sc.SequenceNumber == "" {
		itInput.ShardIteratorType = aws.String(kinesis.ShardIteratorTypeLatest)
	} else {
		itInput.ShardIteratorType = aws.String(kinesis.ShardIteratorTypeAfterSequenceNumber)
		itInput.StartingSequenceNumber = aws.String(sc.SequenceNumber)
	}

	if sc.throughputExceededRetry(func() error {
		itOutput, err = sc.source.client.GetShardIterator(itInput)
		return err
	}) != nil {
		return err
	}

	sc.Iterator = aws.StringValue(itOutput.ShardIterator)

	return nil
}

func (sc *shardConsumer) GetRecords() ([]*kinesis.Record, error) {
	var (
		err      error
		grOutput *kinesis.GetRecordsOutput
	)

	grInput := &kinesis.GetRecordsInput{
		Limit:         aws.Int64(10000),
		ShardIterator: aws.String(sc.Iterator),
	}
	if err := sc.throughputExceededRetry(func() error {
		grOutput, err = sc.source.client.GetRecords(grInput)
		return err
	}); err != nil {
		return nil, err
	}

	sc.Iterator = aws.StringValue(grOutput.NextShardIterator)

	if n := len(grOutput.Records); n > 0 {
		sc.SequenceNumber = aws.StringValue(grOutput.Records[n-1].SequenceNumber)
	}

	return grOutput.Records, nil
}

func (sc *shardConsumer) consume() {
	defer sc.source.wg.Done()
	defer sc.source.cancel()

	err := sc.GetShardIterator()
	if err != nil {
		return
	}

	next := time.NewTimer(100*time.Millisecond - time.Duration(rand.Int63n(int64(50*time.Millisecond))))
	defer next.Stop()
	for sc.Iterator != "" && sc.source.ctx.Err() == nil {

		<-next.C
		resetTo(next, 1*time.Second-time.Duration(rand.Int63n(int64(500*time.Millisecond))))

		records, err := sc.GetRecords()
		if err != nil {
			if ae, ok := err.(awserr.Error); ok && ae.Code() == "ExpiredIteratorException" {
				err := sc.GetShardIterator()
				if err != nil {
					// This error has been passed along to the package user
					// already by throughputExceededRetry.
					return
				}
				continue
			}

			// We don't know what error this is - back off a lot in case it's
			// a bad one!
			resetTo(next, 10*time.Second-time.Duration(rand.Int63n(int64(5*time.Second))))
			continue
		}

		for _, record := range records {
			var ts api.TransactionSet
			err := proto.Unmarshal(record.Data, &ts)
			if err != nil {
				sc.source.reportErr(err)
				return
			}
			for _, tx := range ts.Transaction {
				sc.source.txs <- tx
			}
		}
	}
}

func resetTo(t *time.Timer, d time.Duration) {
	t.Stop()
	select {
	case <-t.C:
	default:
	}
	t.Reset(d)
}

func (s *simpleKinesisTxSource) Transactions() <-chan *api.Transaction {
	return s.txs
}

func (s *simpleKinesisTxSource) Errors() <-chan error {
	return s.errors
}

func (s *simpleKinesisTxSource) Stop() {
	s.cancel()
	s.wg.Wait()
	s.closeOnce.Do(func() {
		close(s.txs)
		close(s.errors)
	})
}
