package common

import (
	"context"
	"errors"
	"fmt"
	"time"

	eventbus "code.justin.tv/eventbus/client"
	"code.justin.tv/eventbus/client/subscriber/sqsclient"
	"code.justin.tv/eventbus/controlplane/e2e/internal/e2eutil"
	"code.justin.tv/eventbus/controlplane/e2e/internal/httpserver"
	"code.justin.tv/eventbus/controlplane/e2e/internal/resource"
	"code.justin.tv/eventbus/controlplane/e2e/internal/test"
	"code.justin.tv/eventbus/controlplane/internal/e2eaccounts"
	"code.justin.tv/eventbus/schema/pkg/clock"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
)

var _ test.TestRunner = &ClientSubscriberTest{}

type ClientSubscriberTest struct {
	test.TestRunner
	AWSConfig    *aws.Config
	Event        eventbus.Message
	Subscription *resource.Subscription
}

func (t *ClientSubscriberTest) TestName() string {
	return fmt.Sprintf("ClientSubscriberTest{%s,%s,%s,%s}", t.Subscription.EventType, t.Subscription.Environment, t.Subscription.TargetName, t.Subscription.ServiceCatalogID)
}

func (t *ClientSubscriberTest) Test(ctx context.Context) {
	ctx = e2eutil.AppendTestPath(ctx, t.TestName())
	t.Log(ctx, "Checking subscriber can receive published event")

	// Get the target to have the QueueURL handy
	target, err := httpserver.Target(t.Subscription.ServiceCatalogID, t.Subscription.TargetName, e2eutil.JobID(ctx))
	if err != nil {
		t.Error(ctx, "Could not retrieve SQS queue URL for eventbus subscriber client", err)
		return // can't continue without a queue URL
	}
	queueURL := target.GetSqs().GetQueueUrl()

	// make a channel to listen for success on
	received := make(chan eventbus.Message)

	// The handler for the event should just push into the received channel
	mux := eventbus.NewMux()

	// If you want to work with a new event from the schema repository in this test, add a case for the event
	// type on this switch statement.
	switch t.Event.EventBusName() {
	case clock.UpdateEventType:
		clock.RegisterUpdateHandler(mux, func(ctx context.Context, h *eventbus.Header, event *clock.Update) error {
			received <- event
			return nil
		})
	default:
		t.Error(ctx, "Invalid event specified for eventbus subscriber client", errors.New("unsuported event type in client subscribe test"))
		return //cannot continue
	}

	sess := session.Must(session.NewSession(e2eaccounts.AccountCredentials().SubscriberConfig()))
	s, err := sqsclient.New(sqsclient.Config{
		Session:    sess,
		Dispatcher: mux.Dispatcher(),
		QueueURL:   queueURL,
	})
	if err != nil {
		t.Error(ctx, "Could not create eventbus subscriber client", err)
		return // cannot continue test
	}
	defer func() {
		if err := s.Shutdown(); err != nil {
			t.Error(ctx, "Could not shutdown eventbus susbcriber client", err)
		}
	}()

	// wait for the event to come through
	select {
	case <-time.After(60 * time.Second):
		t.Error(ctx, "Could not receive event", fmt.Errorf("timed out waiting for ReceiveMessage"))
		return // cannot continue test
	case receivedEvent := <-received:
		// Check the event matches
		if receivedEvent.String() != t.Event.String() {
			t.Error(ctx, "Invalid event received by subscriber", errors.New("received event does not match expected"))
		}
	}
}
