package tests

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

	"floci-sdk-test-go/internal/testutil"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/pipes"
	pipestypes "github.com/aws/aws-sdk-go-v2/service/pipes/types"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
	sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

const (
	pipesAccountID = "000000000000"
	pipesRegion    = "us-east-1"
	pipesRoleArn   = "arn:aws:iam::000000000000:role/pipe-role"
)

func sqsArn(queueName string) string {
	return "arn:aws:sqs:" + pipesRegion + ":" + pipesAccountID + ":" + queueName
}

func TestPipes(t *testing.T) {
	ctx := context.Background()
	pipesSvc := testutil.PipesClient()
	sqsSvc := testutil.SQSClient()

	t.Run("CreatePipe", func(t *testing.T) {
		pipeName := "go-test-create-pipe"
		srcQueue := "go-pipe-src-create"
		tgtQueue := "go-pipe-tgt-create"

		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(srcQueue)})
		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(tgtQueue)})

		t.Cleanup(func() {
			pipesSvc.DeletePipe(ctx, &pipes.DeletePipeInput{Name: aws.String(pipeName)})
			cleanupQueue(ctx, sqsSvc, srcQueue)
			cleanupQueue(ctx, sqsSvc, tgtQueue)
		})

		r, err := pipesSvc.CreatePipe(ctx, &pipes.CreatePipeInput{
			Name:         aws.String(pipeName),
			Source:       aws.String(sqsArn(srcQueue)),
			Target:       aws.String(sqsArn(tgtQueue)),
			RoleArn:      aws.String(pipesRoleArn),
			DesiredState: pipestypes.RequestedPipeStateStopped,
		})
		require.NoError(t, err)
		assert.Equal(t, pipestypes.PipeStateStopped, r.CurrentState)
		assert.True(t, strings.Contains(aws.ToString(r.Arn), pipeName))
	})

	t.Run("DescribePipe", func(t *testing.T) {
		pipeName := "go-test-describe-pipe"
		srcQueue := "go-pipe-src-describe"
		tgtQueue := "go-pipe-tgt-describe"

		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(srcQueue)})
		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(tgtQueue)})

		t.Cleanup(func() {
			pipesSvc.DeletePipe(ctx, &pipes.DeletePipeInput{Name: aws.String(pipeName)})
			cleanupQueue(ctx, sqsSvc, srcQueue)
			cleanupQueue(ctx, sqsSvc, tgtQueue)
		})

		pipesSvc.CreatePipe(ctx, &pipes.CreatePipeInput{
			Name:         aws.String(pipeName),
			Source:       aws.String(sqsArn(srcQueue)),
			Target:       aws.String(sqsArn(tgtQueue)),
			RoleArn:      aws.String(pipesRoleArn),
			DesiredState: pipestypes.RequestedPipeStateStopped,
		})

		r, err := pipesSvc.DescribePipe(ctx, &pipes.DescribePipeInput{Name: aws.String(pipeName)})
		require.NoError(t, err)
		assert.Equal(t, pipeName, aws.ToString(r.Name))
		assert.Equal(t, sqsArn(srcQueue), aws.ToString(r.Source))
		assert.Equal(t, sqsArn(tgtQueue), aws.ToString(r.Target))
		assert.Equal(t, pipestypes.PipeStateStopped, r.CurrentState)
	})

	t.Run("ListPipes", func(t *testing.T) {
		pipeName := "go-test-list-pipe"
		srcQueue := "go-pipe-src-list"
		tgtQueue := "go-pipe-tgt-list"

		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(srcQueue)})
		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(tgtQueue)})

		t.Cleanup(func() {
			pipesSvc.DeletePipe(ctx, &pipes.DeletePipeInput{Name: aws.String(pipeName)})
			cleanupQueue(ctx, sqsSvc, srcQueue)
			cleanupQueue(ctx, sqsSvc, tgtQueue)
		})

		pipesSvc.CreatePipe(ctx, &pipes.CreatePipeInput{
			Name:         aws.String(pipeName),
			Source:       aws.String(sqsArn(srcQueue)),
			Target:       aws.String(sqsArn(tgtQueue)),
			RoleArn:      aws.String(pipesRoleArn),
			DesiredState: pipestypes.RequestedPipeStateStopped,
		})

		r, err := pipesSvc.ListPipes(ctx, &pipes.ListPipesInput{})
		require.NoError(t, err)

		found := false
		for _, p := range r.Pipes {
			if aws.ToString(p.Name) == pipeName {
				found = true
				break
			}
		}
		assert.True(t, found, "pipe should appear in list")
	})

	t.Run("UpdatePipe", func(t *testing.T) {
		pipeName := "go-test-update-pipe"
		srcQueue := "go-pipe-src-update"
		tgtQueue := "go-pipe-tgt-update"

		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(srcQueue)})
		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(tgtQueue)})

		t.Cleanup(func() {
			pipesSvc.DeletePipe(ctx, &pipes.DeletePipeInput{Name: aws.String(pipeName)})
			cleanupQueue(ctx, sqsSvc, srcQueue)
			cleanupQueue(ctx, sqsSvc, tgtQueue)
		})

		pipesSvc.CreatePipe(ctx, &pipes.CreatePipeInput{
			Name:         aws.String(pipeName),
			Source:       aws.String(sqsArn(srcQueue)),
			Target:       aws.String(sqsArn(tgtQueue)),
			RoleArn:      aws.String(pipesRoleArn),
			DesiredState: pipestypes.RequestedPipeStateStopped,
		})

		_, err := pipesSvc.UpdatePipe(ctx, &pipes.UpdatePipeInput{
			Name:         aws.String(pipeName),
			RoleArn:      aws.String(pipesRoleArn),
			Description:  aws.String("updated via SDK"),
			DesiredState: pipestypes.RequestedPipeStateStopped,
		})
		require.NoError(t, err)

		r, err := pipesSvc.DescribePipe(ctx, &pipes.DescribePipeInput{Name: aws.String(pipeName)})
		require.NoError(t, err)
		assert.Equal(t, "updated via SDK", aws.ToString(r.Description))
	})

	t.Run("DeletePipe", func(t *testing.T) {
		pipeName := "go-test-delete-pipe"
		srcQueue := "go-pipe-src-delete"
		tgtQueue := "go-pipe-tgt-delete"

		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(srcQueue)})
		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(tgtQueue)})

		t.Cleanup(func() {
			cleanupQueue(ctx, sqsSvc, srcQueue)
			cleanupQueue(ctx, sqsSvc, tgtQueue)
		})

		pipesSvc.CreatePipe(ctx, &pipes.CreatePipeInput{
			Name:         aws.String(pipeName),
			Source:       aws.String(sqsArn(srcQueue)),
			Target:       aws.String(sqsArn(tgtQueue)),
			RoleArn:      aws.String(pipesRoleArn),
			DesiredState: pipestypes.RequestedPipeStateStopped,
		})

		_, err := pipesSvc.DeletePipe(ctx, &pipes.DeletePipeInput{Name: aws.String(pipeName)})
		require.NoError(t, err)

		_, err = pipesSvc.DescribePipe(ctx, &pipes.DescribePipeInput{Name: aws.String(pipeName)})
		require.Error(t, err)
	})

	t.Run("DescribeNonExistentPipe", func(t *testing.T) {
		_, err := pipesSvc.DescribePipe(ctx, &pipes.DescribePipeInput{Name: aws.String("nonexistent-pipe")})
		require.Error(t, err)
	})

	t.Run("StartAndStopPipe", func(t *testing.T) {
		pipeName := "go-test-startstop-pipe"
		srcQueue := "go-pipe-src-startstop"
		tgtQueue := "go-pipe-tgt-startstop"

		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(srcQueue)})
		sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(tgtQueue)})

		t.Cleanup(func() {
			pipesSvc.DeletePipe(ctx, &pipes.DeletePipeInput{Name: aws.String(pipeName)})
			cleanupQueue(ctx, sqsSvc, srcQueue)
			cleanupQueue(ctx, sqsSvc, tgtQueue)
		})

		pipesSvc.CreatePipe(ctx, &pipes.CreatePipeInput{
			Name:         aws.String(pipeName),
			Source:       aws.String(sqsArn(srcQueue)),
			Target:       aws.String(sqsArn(tgtQueue)),
			RoleArn:      aws.String(pipesRoleArn),
			DesiredState: pipestypes.RequestedPipeStateStopped,
		})

		startR, err := pipesSvc.StartPipe(ctx, &pipes.StartPipeInput{Name: aws.String(pipeName)})
		require.NoError(t, err)
		assert.Equal(t, pipestypes.PipeStateRunning, startR.CurrentState)

		stopR, err := pipesSvc.StopPipe(ctx, &pipes.StopPipeInput{Name: aws.String(pipeName)})
		require.NoError(t, err)
		assert.Equal(t, pipestypes.PipeStateStopped, stopR.CurrentState)
	})

	t.Run("SQSToSQSForwarding", func(t *testing.T) {
		pipeName := "go-test-fwd-pipe"
		srcQueue := "go-pipe-src-fwd"
		tgtQueue := "go-pipe-tgt-fwd"

		srcResp, _ := sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(srcQueue)})
		srcURL := aws.ToString(srcResp.QueueUrl)
		tgtResp, _ := sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(tgtQueue)})
		tgtURL := aws.ToString(tgtResp.QueueUrl)

		t.Cleanup(func() {
			pipesSvc.DeletePipe(ctx, &pipes.DeletePipeInput{Name: aws.String(pipeName)})
			sqsSvc.DeleteQueue(ctx, &sqs.DeleteQueueInput{QueueUrl: aws.String(srcURL)})
			sqsSvc.DeleteQueue(ctx, &sqs.DeleteQueueInput{QueueUrl: aws.String(tgtURL)})
		})

		pipesSvc.CreatePipe(ctx, &pipes.CreatePipeInput{
			Name:         aws.String(pipeName),
			Source:       aws.String(sqsArn(srcQueue)),
			Target:       aws.String(sqsArn(tgtQueue)),
			RoleArn:      aws.String(pipesRoleArn),
			DesiredState: pipestypes.RequestedPipeStateRunning,
		})

		sqsSvc.SendMessage(ctx, &sqs.SendMessageInput{
			QueueUrl:    aws.String(srcURL),
			MessageBody: aws.String("hello from pipes"),
		})

		found := false
		for i := 0; i < 15; i++ {
			time.Sleep(1 * time.Second)
			r, err := sqsSvc.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
				QueueUrl:            aws.String(tgtURL),
				MaxNumberOfMessages: 1,
				WaitTimeSeconds:     1,
			})
			if err == nil && len(r.Messages) > 0 && strings.Contains(aws.ToString(r.Messages[0].Body), "hello from pipes") {
				found = true
				break
			}
		}
		assert.True(t, found, "target queue should receive forwarded message")
	})

	t.Run("FilterCriteriaFiltersMessages", func(t *testing.T) {
		pipeName := "go-test-filter-pipe"
		srcQueue := "go-pipe-src-filter"
		tgtQueue := "go-pipe-tgt-filter"

		srcResp, _ := sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(srcQueue)})
		srcURL := aws.ToString(srcResp.QueueUrl)
		tgtResp, _ := sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(tgtQueue)})
		tgtURL := aws.ToString(tgtResp.QueueUrl)

		t.Cleanup(func() {
			pipesSvc.DeletePipe(ctx, &pipes.DeletePipeInput{Name: aws.String(pipeName)})
			sqsSvc.DeleteQueue(ctx, &sqs.DeleteQueueInput{QueueUrl: aws.String(srcURL)})
			sqsSvc.DeleteQueue(ctx, &sqs.DeleteQueueInput{QueueUrl: aws.String(tgtURL)})
		})

		pipesSvc.CreatePipe(ctx, &pipes.CreatePipeInput{
			Name:         aws.String(pipeName),
			Source:       aws.String(sqsArn(srcQueue)),
			Target:       aws.String(sqsArn(tgtQueue)),
			RoleArn:      aws.String(pipesRoleArn),
			DesiredState: pipestypes.RequestedPipeStateRunning,
			SourceParameters: &pipestypes.PipeSourceParameters{
				FilterCriteria: &pipestypes.FilterCriteria{
					Filters: []pipestypes.Filter{
						{Pattern: aws.String(`{"body": {"status": ["active"]}}`)},
					},
				},
			},
		})

		sqsSvc.SendMessage(ctx, &sqs.SendMessageInput{
			QueueUrl:    aws.String(srcURL),
			MessageBody: aws.String(`{"status": "active", "id": "match-1"}`),
		})
		sqsSvc.SendMessage(ctx, &sqs.SendMessageInput{
			QueueUrl:    aws.String(srcURL),
			MessageBody: aws.String(`{"status": "inactive", "id": "no-match"}`),
		})

		found := false
		for i := 0; i < 15; i++ {
			time.Sleep(1 * time.Second)
			r, err := sqsSvc.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
				QueueUrl:            aws.String(tgtURL),
				MaxNumberOfMessages: 10,
				WaitTimeSeconds:     1,
			})
			if err == nil && len(r.Messages) > 0 {
				hasMatch := false
				hasNoMatch := false
				for _, m := range r.Messages {
					if strings.Contains(aws.ToString(m.Body), "match-1") {
						hasMatch = true
					}
					if strings.Contains(aws.ToString(m.Body), "no-match") {
						hasNoMatch = true
					}
				}
				if hasMatch {
					assert.False(t, hasNoMatch, "non-matching message should not be forwarded")
					found = true
					break
				}
			}
		}
		assert.True(t, found, "target queue should receive matching message")

		attrResp, err := sqsSvc.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{
			QueueUrl:       aws.String(srcURL),
			AttributeNames: []sqstypes.QueueAttributeName{sqstypes.QueueAttributeNameApproximateNumberOfMessages},
		})
		require.NoError(t, err)
		assert.Equal(t, "0", attrResp.Attributes[string(sqstypes.QueueAttributeNameApproximateNumberOfMessages)])
	})

	t.Run("BatchSizeInSourceParameters", func(t *testing.T) {
		pipeName := "go-test-batch-pipe"
		srcQueue := "go-pipe-src-batch"
		tgtQueue := "go-pipe-tgt-batch"

		srcResp, _ := sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(srcQueue)})
		srcURL := aws.ToString(srcResp.QueueUrl)
		tgtResp, _ := sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(tgtQueue)})
		tgtURL := aws.ToString(tgtResp.QueueUrl)

		t.Cleanup(func() {
			pipesSvc.DeletePipe(ctx, &pipes.DeletePipeInput{Name: aws.String(pipeName)})
			sqsSvc.DeleteQueue(ctx, &sqs.DeleteQueueInput{QueueUrl: aws.String(srcURL)})
			sqsSvc.DeleteQueue(ctx, &sqs.DeleteQueueInput{QueueUrl: aws.String(tgtURL)})
		})

		pipesSvc.CreatePipe(ctx, &pipes.CreatePipeInput{
			Name:         aws.String(pipeName),
			Source:       aws.String(sqsArn(srcQueue)),
			Target:       aws.String(sqsArn(tgtQueue)),
			RoleArn:      aws.String(pipesRoleArn),
			DesiredState: pipestypes.RequestedPipeStateRunning,
			SourceParameters: &pipestypes.PipeSourceParameters{
				SqsQueueParameters: &pipestypes.PipeSourceSqsQueueParameters{
					BatchSize: aws.Int32(1),
				},
			},
		})

		for i := 1; i <= 3; i++ {
			sqsSvc.SendMessage(ctx, &sqs.SendMessageInput{
				QueueUrl:    aws.String(srcURL),
				MessageBody: aws.String(fmt.Sprintf("batch-msg-%d", i)),
			})
		}

		foundMessages := make(map[string]bool)
		for i := 0; i < 20 && len(foundMessages) < 3; i++ {
			time.Sleep(1 * time.Second)
			r, err := sqsSvc.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
				QueueUrl:            aws.String(tgtURL),
				MaxNumberOfMessages: 10,
				WaitTimeSeconds:     1,
			})
			if err == nil {
				for _, msg := range r.Messages {
					for j := 1; j <= 3; j++ {
						key := fmt.Sprintf("batch-msg-%d", j)
						if strings.Contains(aws.ToString(msg.Body), key) {
							foundMessages[key] = true
						}
					}
				}
			}
		}
		assert.Len(t, foundMessages, 3, "all 3 messages should arrive at target")
	})

	t.Run("StoppedPipeDoesNotForward", func(t *testing.T) {
		pipeName := "go-test-nofwd-pipe"
		srcQueue := "go-pipe-src-nofwd"
		tgtQueue := "go-pipe-tgt-nofwd"

		srcResp, _ := sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(srcQueue)})
		srcURL := aws.ToString(srcResp.QueueUrl)
		tgtResp, _ := sqsSvc.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(tgtQueue)})
		tgtURL := aws.ToString(tgtResp.QueueUrl)

		t.Cleanup(func() {
			pipesSvc.DeletePipe(ctx, &pipes.DeletePipeInput{Name: aws.String(pipeName)})
			sqsSvc.DeleteQueue(ctx, &sqs.DeleteQueueInput{QueueUrl: aws.String(srcURL)})
			sqsSvc.DeleteQueue(ctx, &sqs.DeleteQueueInput{QueueUrl: aws.String(tgtURL)})
		})

		pipesSvc.CreatePipe(ctx, &pipes.CreatePipeInput{
			Name:         aws.String(pipeName),
			Source:       aws.String(sqsArn(srcQueue)),
			Target:       aws.String(sqsArn(tgtQueue)),
			RoleArn:      aws.String(pipesRoleArn),
			DesiredState: pipestypes.RequestedPipeStateStopped,
		})

		sqsSvc.SendMessage(ctx, &sqs.SendMessageInput{
			QueueUrl:    aws.String(srcURL),
			MessageBody: aws.String("should not forward"),
		})

		time.Sleep(3 * time.Second)

		r, _ := sqsSvc.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
			QueueUrl:            aws.String(tgtURL),
			MaxNumberOfMessages: 1,
			WaitTimeSeconds:     1,
		})
		assert.Empty(t, r.Messages, "target queue should be empty")
	})
}

func cleanupQueue(ctx context.Context, sqsSvc *sqs.Client, queueName string) {
	r, err := sqsSvc.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{QueueName: aws.String(queueName)})
	if err == nil {
		sqsSvc.DeleteQueue(ctx, &sqs.DeleteQueueInput{QueueUrl: r.QueueUrl})
	}
}