package sqspoller

import (
	"context"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/sqs"
	"github.com/pkg/errors"
)

func (c *SQSPoller) newPoller(id int) *poller {
	return &poller{
		id:     id,
		client: c,
		stop:   newStop(c.stopPollers.Context),
	}
}

type poller struct {
	id     int
	client *SQSPoller
	stop   stop
}

func (p *poller) poll() {
	defer p.stop.stop() // Make sure we clean up even in error shutdown
	stopRequests := p.client.stopRequests.Done()
	for running(p.stop.Context) {
		result, err := p.receiveMessages()
		if err != nil {
			// TODO(EDGEPLAT-1901): notify stats hub about errors
			continue
		}

		beginDeliver := time.Now()
		for _, message := range result.Messages {
			select {
			case p.client.deliver <- message:
			case <-stopRequests:
				return
			}
		}

		// TODO(EDGEPLAT-1901): notify stats hub about delivery time and quantity received
		p.client.callback(&HookEventBatchDelivered{
			Messages:    result.Messages,
			DeliveredIn: time.Since(beginDeliver),
		})

	}
}

func (p *poller) receiveMessages() (*sqs.ReceiveMessageOutput, error) {
	ctx, cancel := context.WithTimeout(p.client.stopRequests.Context, receiveMessageMaxWaitTime)
	defer cancel()
	req := &sqs.ReceiveMessageInput{
		QueueUrl:            aws.String(p.client.url),
		MaxNumberOfMessages: aws.Int64(maxMessagesPerPoll),
		VisibilityTimeout:   aws.Int64(p.client.visibilitySeconds),
		WaitTimeSeconds:     aws.Int64(int64(receiveWaitTime.Seconds())),
		AttributeNames: []*string{
			aws.String("ApproximateReceiveCount"),
			aws.String("SentTimestamp"),
			aws.String("ApproximateFirstReceiveTimestamp"),
		},
	}
	result, err := p.client.client.ReceiveMessageWithContext(ctx, req)
	if err != nil {
		err = errors.Wrap(err, "could not poll SQS")
		p.client.callback(&HookEventPollError{
			PollerID: p.id,
			Err:      err,
		})
		return nil, err
	}
	return result, nil
}

func running(ctx context.Context) bool {
	select {
	case <-ctx.Done():
		return false
	default:
		return true
	}
}
