package e2e

import (
	"context"
	"fmt"
	"testing"
	"time"

	"code.justin.tv/eventbus/controlplane/internal/db"
	"code.justin.tv/eventbus/controlplane/internal/db/postgres"
	"code.justin.tv/eventbus/controlplane/internal/twirperr"
	"github.com/pkg/errors"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/suite"
)

type EventStreamsE2ETest struct {
	suite.Suite

	testEventTypeForeignKeys [2]int
	dbConn                   *postgres.PostgresDB
}

func (e *EventStreamsE2ETest) SetupTest() {
	conn, err := newLocalDB()
	e.NoError(err, "no db connection created in EventStreamsE2ETest")
	e.dbConn = conn

	testEventID1, err := e.dbConn.EventTypeCreate(context.Background(), createEventType("TestEvent1", "test event 1", "", "path/to/testevent1.proto"))
	e.NoError(err)

	testEventID2, err := e.dbConn.EventTypeCreate(context.Background(), createEventType("TestEvent2", "test event 2", "", "path/to/testevent2.proto"))
	e.NoError(err)

	e.testEventTypeForeignKeys = [2]int{testEventID1, testEventID2}
}

func (e *EventStreamsE2ETest) TearDownTest() {
	writer := e.dbConn.WriterConn()
	_, err := writer.Exec(fmt.Sprintf("DELETE FROM %s;", postgres.EventStreamsTableName))
	e.NoError(err)

	_, err = writer.Exec(fmt.Sprintf("DELETE FROM %s;", postgres.EventTypesTableName))
	e.NoError(err)

	err = e.dbConn.Close()
	e.NoError(err)
}

func (e *EventStreamsE2ETest) TestEventStreamsE2E() {
	t := e.T()
	ctx := context.Background()

	eventStream1 := createEventStream("production", e.testEventTypeForeignKeys[0])
	id1, err := e.dbConn.EventStreamCreate(ctx, eventStream1)
	e.NoError(err)
	e.False(eventStream1.Deprecated)
	eventStream1.ID = id1

	eventStream2 := createEventStream("production", e.testEventTypeForeignKeys[1])
	id2, err := e.dbConn.EventStreamCreate(ctx, eventStream2)
	e.NoError(err)
	e.False(eventStream2.Deprecated)
	eventStream2.ID = id2

	eventStream3 := createEventStream("staging", e.testEventTypeForeignKeys[0])
	id3, err := e.dbConn.EventStreamCreate(ctx, eventStream3)
	e.NoError(err)
	e.False(eventStream3.Deprecated)
	eventStream3.ID = id3

	eventStreamsStaging, err := e.dbConn.EventStreamsByEnvironment(ctx, "staging")
	e.NoError(err)
	e.Equalf(1, len(eventStreamsStaging), "expected %d staging event streams, but got %d", 1, len(eventStreamsStaging))
	assertEventStreamEquality(t, eventStream3, eventStreamsStaging[0])

	eventStreamsProd, err := e.dbConn.EventStreamsByEnvironment(ctx, "production")
	e.NoError(err)
	e.Equalf(2, len(eventStreamsProd), "expected %d prod event streams, but got %d", 2, len(eventStreamsProd))

	// in case our event streams are received in different orders from the db,
	// we make an explicit check for the id
	if eventStreamsProd[0].ID == id1 {
		assertEventStreamEquality(t, eventStream1, eventStreamsProd[0])
		assertEventStreamEquality(t, eventStream2, eventStreamsProd[1])
	} else if eventStreamsProd[0].ID == id2 {
		assertEventStreamEquality(t, eventStream1, eventStreamsProd[1])
		assertEventStreamEquality(t, eventStream2, eventStreamsProd[0])
	}

	allEventStreams, err := e.dbConn.EventStreams(ctx)
	e.NoError(err)
	e.Equalf(3, len(allEventStreams), "expected %d total event streams, but got %d", 3, len(allEventStreams))

	eventStreamByID1, err := e.dbConn.EventStreamByID(ctx, id1)
	e.NoError(err)
	assertEventStreamEquality(t, eventStream1, eventStreamByID1)
	e.False(eventStreamByID1.Deprecated)

	notFound1, err := e.dbConn.EventStreamByID(ctx, 999)
	e.True(errors.Cause(err).(twirperr.DBError).EventStreamNotFound())
	e.Nil(notFound1)

	notFound2, err := e.dbConn.EventStreamByNameAndEnvironment(ctx, "DoesNotExist", "DoesNotExist")
	e.True(errors.Cause(err).(twirperr.DBError).EventStreamNotFound())
	e.Nil(notFound2)

	_, err = e.dbConn.EventStreamUpdate(ctx, 666, &db.EventStreamEditable{})
	e.Equal(err, db.ErrNotSupported)

	fakeLease := &postgres.PostgresAWSLease{
		ExpiresAt: time.Now().Add(1 * time.Minute),
	}
	_, err = e.dbConn.EventStreamUpdateInfra(ctx, fakeLease, 666, &db.EventStreamInfraUpdate{})
	e.Equal(err, db.ErrResourceNotFound)
}

func TestEventStreamsE2E(T *testing.T) {
	suite.Run(T, &EventStreamsE2ETest{})
}

func createEventStream(env string, eventTypeID int) *db.EventStream {
	return &db.EventStream{
		Environment: env,
		EventTypeID: eventTypeID,
	}
}

func assertEventStreamEquality(t *testing.T, es1, es2 *db.EventStream) {
	assert.Equal(t, es1.ID, es2.ID)
	assert.Equal(t, es1.Environment, es2.Environment)
	assert.Equal(t, es1.EventTypeID, es2.EventTypeID)
}
