package sqspoller

import (
	"context"
	"sync"

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

type HookCallback func(HookEvent)

type SQSPoller struct {
	client            sqsiface.SQSAPI
	url               string
	VisibilitySeconds int64
	callback          HookCallback

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

	deliver chan *sqs.Message
	acks    chan string
}

func New(client sqsiface.SQSAPI, queueURL string, visibilitySeconds int64, numPollers int, deliver chan *sqs.Message, callback HookCallback) *SQSPoller {
	c := &SQSPoller{
		client:            client,
		url:               queueURL,
		VisibilitySeconds: visibilitySeconds,
		callback:          callback,

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

		deliver: deliver,
		acks:    make(chan string),
	}

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

	runTask(&c.hubGroup, c.ackHub)

	return c
}

// Close this SQS client in stages, shutting down all running processes.
//
// This method will block until all running goroutines have completed.
func (c *SQSPoller) StopPollers() {
	c.stopPollers.stop()
	c.pollerGroup.Wait()
	//	close(c.deliver)
	//	c.deliverGroup.Wait()
}

func (c *SQSPoller) StopAcking() {
	close(c.acks)
	c.hubGroup.Wait()
}

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()
	}()
}
