package sqspoller

import (
	"context"
	"sync"

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

type SQSPoller struct {
	client            SQSAPI
	url               string
	visibilitySeconds int64
	callback          HookCallback

	stopPollers  stop
	pollerGroup  sync.WaitGroup
	stopRequests stop
	hubGroup     sync.WaitGroup

	deliver chan<- *sqs.Message
	acks    chan string
}

// New creates a new SQSPoller that will immediately start polling SQS.
func New(config Config) *SQSPoller {
	config.defaults()

	c := &SQSPoller{
		client:            config.Client,
		url:               config.QueueURL,
		visibilitySeconds: config.VisibilitySeconds,
		callback:          config.HookCallback,

		stopPollers:  newStop(context.Background()),
		stopRequests: newStop(context.Background()),

		deliver: config.Deliver,
		acks:    make(chan string),
	}

	for i := 0; i < config.NumPollers; i++ {
		poller := c.newPoller(i)
		runTask(&c.pollerGroup, poller.poll)
	}

	runTask(&c.hubGroup, c.ackHub)

	return c
}

// StopPollers is shutdown stage1; it will stop the polling goroutines.
//
// The client will continue to process any in-flight messages until
// StopAcking is called.
func (c *SQSPoller) StopPollers() {
	c.stopPollers.stop()
	c.pollerGroup.Wait()
}

// StopAcking is shutdown stage2; it will stop the acks being sent.
func (c *SQSPoller) StopAcking() {
	close(c.acks)
	c.hubGroup.Wait()
}

// CancelOutgoingRequests is a way to cancel all outgoing requests to potentially speed up shutdown.
func (c *SQSPoller) CancelOutgoingRequests() {
	c.stopRequests.stop()
}

type stop struct {
	context.Context
	stop func()
}

func newStop(base context.Context) stop {
	ctx, cancel := context.WithCancel(base)
	return stop{
		Context: ctx,
		stop:    cancel,
	}
}

func runTask(wg *sync.WaitGroup, f func()) {
	wg.Add(1)
	go func() {
		defer wg.Done()
		f()
	}()
}
