// package lambdafunc implements Lambda function handler support for the event bus.
//
// This package returns a function designed to be registered with the
// aws-lambda-go library: https://github.com/aws/aws-lambda-go
//
// When the lambda is invoked, it handles event payloads, parsing them and
// forwarding them to your dispatcher.
package lambdafunc

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"

	"github.com/aws/aws-lambda-go/events"
	"github.com/pkg/errors"

	eventbus "code.justin.tv/eventbus/client"
	"code.justin.tv/eventbus/client/internal/retrycount"
	"code.justin.tv/eventbus/client/lowlevel/snsmarshal"
)

type SQSLambdaHandler func(context.Context, events.SQSEvent) error
type SNSLambdaHandler func(context.Context, events.SNSEvent) error

// NewSQS creates an SQS lambda handler, to be used with your SQS target.
//
// The SQS lambda handler may receive more than one event in the payload;
// in this case it will call Dispatch on each event _sequentially_.
// If any event in the payload errors it will return that error immediately
// and not try to process any further events in the payload.
func NewSQS(dispatcher eventbus.Dispatcher) SQSLambdaHandler {
	return func(ctx context.Context, event events.SQSEvent) error {
		for i, record := range event.Records {
			// SNS message JSON is inside the Body of the SQS event
			var entity events.SNSEntity

			err := json.Unmarshal([]byte(record.Body), &entity)
			if err != nil {
				return errors.Wrapf(err, "could not decode SNS entity from SQS body of message %d", i)
			}

			decoded, err := snsmarshal.DecodeString(entity.Message)
			if err != nil {
				return errors.Wrapf(err, "could not decode Event Bus entity from SNS payload %d", i)
			}

			receiveCount, ok := record.Attributes["ApproximateReceiveCount"]
			if ok {
				i, err := strconv.Atoi(receiveCount)
				if err != nil {
					return errors.Wrap(err, "could not parse receive count in SQS attributes")
				}

				ctx = retrycount.Set(ctx, i)
			}
			err = dispatcher.Dispatch(ctx, decoded)
			if err != nil {
				return errors.Wrapf(err, "error dispatching SQS message %d", i)
			}
		}
		return nil
	}
}

// NewSNS creates an SNS lambda handler, used to connect directly to an SNS topic.
func NewSNS(dispatcher eventbus.Dispatcher) SNSLambdaHandler {
	return func(ctx context.Context, event events.SNSEvent) error {
		if len(event.Records) != 1 {
			// While event.Records is an array, we will not receive more than one message
			// as stated by the SNS FAQ under reliability: https://aws.amazon.com/sns/faqs/
			return fmt.Errorf("Invalid record count %d != 1", len(event.Records))
		}

		decoded, err := snsmarshal.DecodeString(event.Records[0].SNS.Message)
		if err != nil {
			return err
		}

		return dispatcher.Dispatch(ctx, decoded)
	}
}
