package messaging

import (
	"context"
	"errors"
	"fmt"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"sync"
	"testing"
	"time"
)

type connectionPoolMock struct {
	mock.Mock
}

func (m *connectionPoolMock) Close() error {
	return m.Called().Error(0)
}

func (m *connectionPoolMock) NewHandle() *ConnectionPoolHandle {
	return m.Called().Get(0).(*ConnectionPoolHandle)
}

func (m *connectionPoolMock) GetConnection(handle *ConnectionPoolHandle) (*AmqpConnection, error) {
	return m.Called(handle).Get(0).(*AmqpConnection), m.Called(handle).Error(1)
}

var _ ConnectionPool = (*connectionPoolMock)(nil)

func Test_NewAmqpMessagingApi(t *testing.T) {
	_, err := NewAmqpApi(
		AmqpConnectionPoolConfig{
			Parameters: AmqpConnectionConfig{BrokerUrl: "not-handled-protocol://localhost:5672"},
			PoolSize:   1,
		})
	assert.EqualError(t, err, "new amqp connection pool: initialize connections: new connection: new internal connection: internal connect: dial: unsupported scheme \"not-handled-protocol\"")
}

func Test_AmqpMessagingApi_Send(t *testing.T) {
	// Specify test timeout
	ctx, cancelFn := context.WithTimeout(context.Background(), 120*time.Second)
	defer cancelFn()

	// Start solace docker container
	solaceContainer, err := NewSolaceContainer(context.Background())
	assert.NoError(t, err)
	defer solaceContainer.Stop()

	t.Run("Missing topic prefix", func(t *testing.T) {
		defer solaceContainer.StopOnError()

		api, err := NewAmqpApi(AmqpConnectionPoolConfig{
			Parameters: AmqpConnectionConfig{BrokerUrl: solaceContainer.AmqpConnectionString},
			PoolSize:   1,
		})
		assert.NoError(t, err)

		err = api.Send(ctx, "topic-name", []byte{}, "application/json", make(map[string]any))
		assert.EqualError(t, err, "send: topic \"topic-name\" name lacks mandatory prefix \"topic://\"\nretry send: topic \"topic-name\" name lacks mandatory prefix \"topic://\"")
	})

	t.Run("send successfully", func(t *testing.T) {
		defer solaceContainer.StopOnError()

		// Initialize the solace queue
		topicSubscriptionTopicPattern := "auditlog/>"
		queueName := "send-successfully"
		assert.NoError(t, solaceContainer.QueueCreate(ctx, queueName))
		assert.NoError(t, solaceContainer.TopicSubscriptionCreate(ctx, queueName, topicSubscriptionTopicPattern))
		topicName := fmt.Sprintf("topic://auditlog/%s", "amqp-send-successfully")
		assert.NoError(t, solaceContainer.ValidateTopicName(topicSubscriptionTopicPattern, topicName))

		api, err := NewDefaultAmqpApi(AmqpConnectionConfig{BrokerUrl: solaceContainer.AmqpConnectionString})
		assert.NoError(t, err)

		data := []byte("data")
		applicationProperties := make(map[string]interface{})
		applicationProperties["key"] = "value"

		err = api.Send(ctx, topicName, data, "application/json", applicationProperties)
		assert.NoError(t, err)

		message, err := solaceContainer.NextMessage(ctx, fmt.Sprintf("queue://%s", queueName), true)
		assert.NoError(t, err)
		assert.Equal(t, "data", string(message.Data[0]))
		assert.Equal(t, topicName, *message.Properties.To)
		assert.Equal(t, "application/json", *message.Properties.ContentType)
		assert.Equal(t, applicationProperties, message.ApplicationProperties)

		err = api.Close(ctx)
		assert.NoError(t, err)
	})
}

func Test_AmqpMessagingApi_Send_Special_Cases(t *testing.T) {

	channelReceiver := func(channel chan struct{}) <-chan struct{} {
		return channel
	}

	newActiveConnection := func() *AmqpConnection {
		channel := make(chan struct{})
		conn := &amqpConnMock{}
		conn.On("Done", mock.Anything).Return(channelReceiver(channel))
		return &AmqpConnection{
			connectionName: "test",
			lock:           sync.RWMutex{},
			conn:           conn,
		}
	}

	newClosedConnection := func() *AmqpConnection {
		channel := make(chan struct{})
		close(channel)

		conn := &amqpConnMock{}
		conn.On("Done", mock.Anything).Return(channelReceiver(channel))
		return &AmqpConnection{
			connectionName: "test",
			lock:           sync.RWMutex{},
			conn:           conn,
		}
	}

	t.Run("connection nil sender nil", func(t *testing.T) {
		sender := &amqpSenderMock{}
		sender.On("Send", mock.Anything, mock.Anything, mock.Anything).Return(nil)

		session := &amqpSessionMock{}
		session.On("NewSender", mock.Anything, mock.Anything, mock.Anything).Return(sender, nil)

		connection := newActiveConnection()
		conn := connection.conn.(*amqpConnMock)
		conn.On("NewSession", mock.Anything, mock.Anything).Return(session, nil)

		pool := &connectionPoolMock{}
		pool.On("GetConnection", mock.Anything).Return(connection, nil)

		amqpApi := &AmqpApi{
			connectionPool:       pool,
			connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
			senderCache:          make(map[string]*AmqpSenderSession),
		}

		err := amqpApi.Send(context.Background(), "topic://some-topic", []byte("data"), "application/json", make(map[string]any))
		assert.NoError(t, err)

		sender.AssertNumberOfCalls(t, "Send", 1)
		session.AssertNumberOfCalls(t, "NewSender", 1)
		pool.AssertNumberOfCalls(t, "GetConnection", 2)
	})

	t.Run("connection closed sender nil", func(t *testing.T) {
		sender := &amqpSenderMock{}
		sender.On("Send", mock.Anything, mock.Anything, mock.Anything).Return(nil)

		session := &amqpSessionMock{}
		session.On("NewSender", mock.Anything, mock.Anything, mock.Anything).Return(sender, nil)

		connection := newActiveConnection()
		conn := connection.conn.(*amqpConnMock)
		conn.On("NewSession", mock.Anything, mock.Anything).Return(session, nil)

		pool := &connectionPoolMock{}
		pool.On("GetConnection", mock.Anything).Return(connection, nil)

		closedConnection := newClosedConnection()
		closedConnMock := closedConnection.conn.(*amqpConnMock)
		amqpApi := &AmqpApi{
			connection:           closedConnection,
			connectionPool:       pool,
			connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
			senderCache:          make(map[string]*AmqpSenderSession),
		}

		err := amqpApi.Send(context.Background(), "topic://some-topic", []byte("data"), "application/json", make(map[string]any))
		assert.NoError(t, err)

		sender.AssertNumberOfCalls(t, "Send", 1)
		session.AssertNumberOfCalls(t, "NewSender", 1)
		pool.AssertNumberOfCalls(t, "GetConnection", 2)
		closedConnMock.AssertNumberOfCalls(t, "Done", 1)
	})

	t.Run("connection nil get connection fail", func(t *testing.T) {
		var connection *AmqpConnection = nil

		pool := &connectionPoolMock{}
		pool.On("GetConnection", mock.Anything).Return(connection, errors.New("connection error"))

		amqpApi := &AmqpApi{
			connectionPool:       pool,
			connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
			senderCache:          make(map[string]*AmqpSenderSession),
		}

		err := amqpApi.Send(context.Background(), "topic://some-topic", []byte("data"), "application/json", make(map[string]any))
		assert.EqualError(t, err, "get connection: connection error")

		pool.AssertNumberOfCalls(t, "GetConnection", 2)
	})

	t.Run("connection active sender nil", func(t *testing.T) {
		sender := &amqpSenderMock{}
		sender.On("Send", mock.Anything, mock.Anything, mock.Anything).Return(nil)

		session := &amqpSessionMock{}
		session.On("NewSender", mock.Anything, mock.Anything, mock.Anything).Return(sender, nil)

		connection := newActiveConnection()
		conn := connection.conn.(*amqpConnMock)
		conn.On("NewSession", mock.Anything, mock.Anything).Return(session, nil)

		amqpApi := &AmqpApi{
			connection:  connection,
			senderCache: make(map[string]*AmqpSenderSession),
		}

		err := amqpApi.Send(context.Background(), "topic://some-topic", []byte("data"), "application/json", make(map[string]any))
		assert.NoError(t, err)

		sender.AssertNumberOfCalls(t, "Send", 1)
		session.AssertNumberOfCalls(t, "NewSender", 1)
	})

	t.Run("connection active new sender fail", func(t *testing.T) {
		var sender *amqpSenderMock = nil

		session := &amqpSessionMock{}
		session.On("NewSender", mock.Anything, mock.Anything, mock.Anything).Return(sender, errors.New("new sender error"))
		session.On("Close", mock.Anything).Return(nil)

		connection := newActiveConnection()
		conn := connection.conn.(*amqpConnMock)
		conn.On("NewSession", mock.Anything, mock.Anything).Return(session, nil)

		amqpApi := &AmqpApi{
			connection:  connection,
			senderCache: make(map[string]*AmqpSenderSession),
		}

		err := amqpApi.Send(context.Background(), "topic://some-topic", []byte("data"), "application/json", make(map[string]any))
		assert.EqualError(t, err, "new sender: new internal sender: new sender error")

		session.AssertNumberOfCalls(t, "NewSender", 1)
		session.AssertNumberOfCalls(t, "Close", 1)
	})

	t.Run("connection active sender set", func(t *testing.T) {
		sender := &amqpSenderMock{}
		sender.On("Send", mock.Anything, mock.Anything, mock.Anything).Return(nil)

		topic := "topic://some-topic"
		amqpApi := &AmqpApi{
			connection:  newActiveConnection(),
			senderCache: map[string]*AmqpSenderSession{topic: {sender: sender}},
		}

		err := amqpApi.Send(context.Background(), topic, []byte("data"), "application/json", make(map[string]any))
		assert.NoError(t, err)

		sender.AssertNumberOfCalls(t, "Send", 1)
	})

	t.Run("send fail", func(t *testing.T) {
		sender := &amqpSenderMock{}
		sender.On("Send", mock.Anything, mock.Anything, mock.Anything).Return(errors.New("send error"))

		session := &amqpSessionMock{}
		session.On("NewSender", mock.Anything, mock.Anything, mock.Anything).Return(sender, nil)

		topic := "topic://some-topic"
		connection := newActiveConnection()
		connection.conn.(*amqpConnMock).On("NewSession", mock.Anything, mock.Anything, mock.Anything).Return(session, nil)
		amqpApi := &AmqpApi{
			connection:  connection,
			senderCache: map[string]*AmqpSenderSession{topic: {sender: sender}},
		}

		err := amqpApi.Send(context.Background(), topic, []byte("data"), "application/json", make(map[string]any))
		assert.EqualError(t, err, "send: send error\nretry send: send error")

		sender.AssertNumberOfCalls(t, "Send", 2)
	})
}

func Test_AmqpMessagingApi_Close(t *testing.T) {

	t.Run("close without cached senders", func(t *testing.T) {
		pool := &connectionPoolMock{}
		pool.On("Close").Return(nil)

		amqpApi := &AmqpApi{
			connectionPool:       pool,
			connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
			senderCache:          make(map[string]*AmqpSenderSession),
		}

		err := amqpApi.Close(context.Background())
		assert.NoError(t, err)

		pool.AssertNumberOfCalls(t, "Close", 1)
	})

	t.Run("close fail without cached senders", func(t *testing.T) {
		pool := &connectionPoolMock{}
		pool.On("Close").Return(errors.New("close error"))

		amqpApi := &AmqpApi{
			connectionPool:       pool,
			connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
			senderCache:          make(map[string]*AmqpSenderSession),
		}

		err := amqpApi.Close(context.Background())
		assert.EqualError(t, err, "close: close pool: close error")

		pool.AssertNumberOfCalls(t, "Close", 1)
	})

	t.Run("close with cached senders", func(t *testing.T) {
		pool := &connectionPoolMock{}
		pool.On("Close").Return(nil)

		session := &amqpSessionMock{}
		session.On("Close", mock.Anything).Return(nil)
		sender := &amqpSenderMock{}
		sender.On("Close", mock.Anything).Return(nil)
		senderSession := &AmqpSenderSession{
			session: session,
			sender:  sender,
		}

		amqpApi := &AmqpApi{
			connectionPool:       pool,
			connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
			senderCache:          map[string]*AmqpSenderSession{"key": senderSession},
		}

		err := amqpApi.Close(context.Background())
		assert.NoError(t, err)
		assert.Equal(t, 0, len(amqpApi.senderCache))

		pool.AssertNumberOfCalls(t, "Close", 1)
		session.AssertNumberOfCalls(t, "Close", 1)
		sender.AssertNumberOfCalls(t, "Close", 1)
	})

	t.Run("close fail with cached senders", func(t *testing.T) {
		pool := &connectionPoolMock{}
		pool.On("Close").Return(nil)

		session := &amqpSessionMock{}
		session.On("Close", mock.Anything).Return(nil)
		sender := &amqpSenderMock{}
		sender.On("Close", mock.Anything).Return(errors.New("close sender error"))
		senderSession := &AmqpSenderSession{
			session: session,
			sender:  sender,
		}

		amqpApi := &AmqpApi{
			connectionPool:       pool,
			connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
			senderCache:          map[string]*AmqpSenderSession{"key": senderSession},
		}

		err := amqpApi.Close(context.Background())
		assert.EqualError(t, err, "close: close session: close sender error")
		assert.Equal(t, 0, len(amqpApi.senderCache))

		pool.AssertNumberOfCalls(t, "Close", 1)
		session.AssertNumberOfCalls(t, "Close", 1)
		sender.AssertNumberOfCalls(t, "Close", 1)
	})

	t.Run("close fail", func(t *testing.T) {
		pool := &connectionPoolMock{}
		pool.On("Close").Return(errors.New("close pool error"))

		session := &amqpSessionMock{}
		session.On("Close", mock.Anything).Return(errors.New("close session error"))
		sender := &amqpSenderMock{}
		sender.On("Close", mock.Anything).Return(errors.New("close sender error"))
		senderSession := &AmqpSenderSession{
			session: session,
			sender:  sender,
		}

		amqpApi := &AmqpApi{
			connectionPool:       pool,
			connectionPoolHandle: &ConnectionPoolHandle{connectionOffset: 0},
			senderCache:          map[string]*AmqpSenderSession{"key": senderSession},
		}

		err := amqpApi.Close(context.Background())
		assert.EqualError(t, err, "close: close session: close sender error\nclose session error\nclose pool: close pool error")
		assert.Equal(t, 0, len(amqpApi.senderCache))

		pool.AssertNumberOfCalls(t, "Close", 1)
		session.AssertNumberOfCalls(t, "Close", 1)
		sender.AssertNumberOfCalls(t, "Close", 1)
	})
}