package kafkatopic_test

import (
	"context"
	"fmt"
	"log"
	"os"
	"regexp"
	"testing"
	"time"

	"github.com/aiven/aiven-go-client/v2"
	"github.com/google/go-cmp/cmp"
	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/retry"
	"github.com/hashicorp/terraform-plugin-testing/helper/acctest"
	"github.com/hashicorp/terraform-plugin-testing/helper/resource"
	"github.com/hashicorp/terraform-plugin-testing/terraform"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	acc "github.com/aiven/terraform-provider-aiven/internal/acctest"
	"github.com/aiven/terraform-provider-aiven/internal/schemautil"
	"github.com/aiven/terraform-provider-aiven/internal/sdkprovider/kafkatopicrepository"
	"github.com/aiven/terraform-provider-aiven/internal/sdkprovider/service/kafkatopic"
)

func TestAccAivenKafkaTopic(t *testing.T) {
	if os.Getenv(resource.EnvTfAcc) != "1" {
		t.Skipf("Set '%s=1' to run this acceptance test", resource.EnvTfAcc)
	}

	orgName := acc.OrganizationName()
	projectName := acc.ProjectName()
	kafkaName := acc.RandName("kafka")

	// Creates shared Kafka
	cleanup, err := acc.CreateTestService(
		t.Context(),
		projectName,
		kafkaName,
		acc.WithServiceType("kafka"),
		acc.WithPlan("startup-4"),
		acc.WithCloud("google-europe-west1"),
	)

	require.NoError(t, err)
	defer cleanup()

	t.Run("basic", func(t *testing.T) {
		defer kafkatopicrepository.ForgetService(projectName, kafkaName)

		resourceName := "aiven_kafka_topic.foo"
		topic2ResourceName := "aiven_kafka_topic.topic2"
		rName := acctest.RandStringFromCharSet(10, acctest.CharSetAlphaNum)
		resource.Test(t, resource.TestCase{
			PreCheck:                 func() { acc.TestAccPreCheck(t) },
			ProtoV6ProviderFactories: acc.TestProtoV6ProviderFactories,
			CheckDestroy:             testAccCheckAivenKafkaTopicResourceDestroy,
			Steps: []resource.TestStep{
				{
					Config: testAccKafkaTopicResource(orgName, projectName, kafkaName, rName),
					Check: resource.ComposeTestCheckFunc(
						testAccCheckAivenKafkaTopicAttributes("data.aiven_kafka_topic.topic"),
						resource.TestCheckResourceAttr(resourceName, "project", projectName),
						resource.TestCheckResourceAttr(resourceName, "service_name", kafkaName),
						resource.TestCheckResourceAttr(resourceName, "topic_name", fmt.Sprintf("test-acc-topic-%s", rName)),
						resource.TestCheckResourceAttr(resourceName, "partitions", "3"),
						resource.TestCheckResourceAttr(resourceName, "replication", "2"),
						resource.TestCheckResourceAttr(resourceName, "termination_protection", "false"),
						resource.TestCheckResourceAttr(resourceName, "config.0.retention_bytes", "1234"),
						resource.TestCheckResourceAttr(resourceName, "config.0.segment_bytes", "1610612736"),
						resource.TestCheckResourceAttr(topic2ResourceName, "topic_description", fmt.Sprintf("test-acc-topic2-desc-%s", rName)),
						resource.TestCheckResourceAttrSet(topic2ResourceName, "owner_user_group_id"),
					),
				},
			},
		})
	})

	// The test proves that TF can create hundreds of topics without hitting any server issues
	t.Run("many_topics", func(t *testing.T) {
		defer kafkatopicrepository.ForgetService(projectName, kafkaName)

		topicName := acc.RandName("topic")
		resource.Test(t, resource.TestCase{
			PreCheck:                 func() { acc.TestAccPreCheck(t) },
			ProtoV6ProviderFactories: acc.TestProtoV6ProviderFactories,
			CheckDestroy:             testAccCheckAivenKafkaTopicResourceDestroy,
			Steps: []resource.TestStep{
				{
					Config: testAccKafka451TopicResource(projectName, kafkaName, topicName),
					Check: resource.ComposeTestCheckFunc(
						func(state *terraform.State) error {
							assert.Len(t, state.RootModule().Resources, 451)
							return nil
						},
					),
				},
			},
		})
	})

	t.Run("termination_protection", func(t *testing.T) {
		defer kafkatopicrepository.ForgetService(projectName, kafkaName)

		topicName := acc.RandName("topic")
		resourceName := "aiven_kafka_topic.foo"
		resource.Test(t, resource.TestCase{
			PreCheck:                 func() { acc.TestAccPreCheck(t) },
			ProtoV6ProviderFactories: acc.TestProtoV6ProviderFactories,
			CheckDestroy:             testAccCheckAivenKafkaTopicResourceDestroy,
			Steps: []resource.TestStep{
				{
					Config:                    testAccKafkaTopicTerminationProtectionResource(projectName, kafkaName, topicName),
					PreventPostDestroyRefresh: true,
					ExpectNonEmptyPlan:        true,
					PlanOnly:                  true,
					Check: resource.ComposeTestCheckFunc(
						resource.TestCheckResourceAttr(resourceName, "project", projectName),
						resource.TestCheckResourceAttr(resourceName, "service_name", kafkaName),
						resource.TestCheckResourceAttr(resourceName, "topic_name", topicName),
						resource.TestCheckResourceAttr(resourceName, "partitions", "3"),
						resource.TestCheckResourceAttr(resourceName, "replication", "2"),
						resource.TestCheckResourceAttr(resourceName, "termination_protection", "true"),
					),
				},
			},
		})
	})

	// TestAccAivenKafkaTopic_recreate validates that topic is recreated if it is missing
	// Kafka loses all topics on turn off/on, then TF recreates topics. This test imitates the case.
	t.Run("recreate_missing", func(t *testing.T) {
		defer kafkatopicrepository.ForgetService(projectName, kafkaName)

		topicResource := "aiven_kafka_topic.topic"
		topicName := acc.RandName("topic")
		kafkaID := fmt.Sprintf("%s/%s", projectName, kafkaName)
		topicID := fmt.Sprintf("%s/%s", kafkaID, topicName)

		config := testAccAivenKafkaTopicResourceRecreateMissing(projectName, kafkaName, topicName)
		resource.Test(t, resource.TestCase{
			PreCheck:                 func() { acc.TestAccPreCheck(t) },
			ProtoV6ProviderFactories: acc.TestProtoV6ProviderFactories,
			CheckDestroy:             testAccCheckAivenKafkaTopicResourceDestroy,
			Steps: []resource.TestStep{
				{
					// Step 1: setups resources, creates the state
					Config: config,
					Check: resource.ComposeTestCheckFunc(
						resource.TestCheckResourceAttr(topicResource, "id", topicID),
					),
				},
				{
					// Step 2: deletes topic, then runs apply, same config & checks
					PreConfig: func() {
						ctx := t.Context()
						client := acc.GetTestAivenClient()

						// deletes
						err := client.KafkaTopics.Delete(ctx, projectName, kafkaName, topicName)
						require.NoError(t, err)

						// Makes sure topic does not exist
						err = retry.RetryContext(ctx, time.Minute, func() *retry.RetryError {
							_, err := client.KafkaTopics.Get(ctx, projectName, kafkaName, topicName)
							if err != nil {
								return &retry.RetryError{
									Err:       fmt.Errorf(`can't get the "missing" topic: %w`, err),
									Retryable: aiven.IsNotFound(err),
								}
							}
							return nil
						})
						require.NoError(t, err)

						// We need to remove it from reps cache
						assert.NoError(t, kafkatopicrepository.ForgetTopic(projectName, kafkaName, topicName))
					},
					// Now plan shows a diff
					ExpectNonEmptyPlan: true,
					RefreshState:       true,
				},
				{
					// Step 3: recreates the topic
					Config: config,
					Check: resource.ComposeTestCheckFunc(
						// Saved in state
						resource.TestCheckResourceAttr(topicResource, "id", topicID),
						func(_ *terraform.State) error {
							// Topic exists and active
							client := acc.GetTestAivenClient()
							return retry.RetryContext(
								t.Context(),
								time.Minute,
								func() *retry.RetryError {
									ctx := t.Context()
									tc, err := client.KafkaTopics.Get(ctx, projectName, kafkaName, topicName)
									if err != nil {
										return &retry.RetryError{
											Err:       fmt.Errorf(`can't get the "missing" topic: %w`, err),
											Retryable: aiven.IsNotFound(err),
										}
									}
									assert.Equal(t, "ACTIVE", tc.State)
									return nil
								},
							)
						},
					),
				},
			},
		})
	})

	t.Run("import_missing", func(t *testing.T) {
		topicName := acc.RandName("topic")
		resource.Test(t, resource.TestCase{
			PreCheck:                 func() { acc.TestAccPreCheck(t) },
			ProtoV6ProviderFactories: acc.TestProtoV6ProviderFactories,
			CheckDestroy:             testAccCheckAivenKafkaTopicResourceDestroy,
			Steps: []resource.TestStep{
				{
					// Tries to import non-existing topic
					// Must fail not create
					Config:        testAccAivenKafkaTopicResourceImportMissing(projectName, kafkaName, topicName),
					ResourceName:  "aiven_kafka_topic.topic",
					ImportState:   true,
					ImportStateId: fmt.Sprintf("%s/%s/%s", projectName, kafkaName, topicName),
					ExpectError:   regexp.MustCompile(`While attempting to import an existing object to "aiven_kafka_topic.topic"`),
				},
			},
		})
	})

	t.Run("conflicts_if_exists", func(t *testing.T) {
		defer kafkatopicrepository.ForgetService(projectName, kafkaName)

		topicName := acc.RandName("topic")
		resource.Test(t, resource.TestCase{
			PreCheck:                 func() { acc.TestAccPreCheck(t) },
			ProtoV6ProviderFactories: acc.TestProtoV6ProviderFactories,
			CheckDestroy:             testAccCheckAivenKafkaTopicResourceDestroy,
			Steps: []resource.TestStep{
				{
					Config:      testAccAivenKafkaTopicConflictsIfExists(projectName, kafkaName, topicName),
					ExpectError: regexp.MustCompile(fmt.Sprintf(`topic conflict, already exists: %q`, topicName)),
				},
			},
		})
	})
}

func testAccKafka451TopicResource(projectName, kafkaName, topicName string) string {
	return fmt.Sprintf(`
resource "aiven_kafka_topic" "foo" {
  count = 451

  project      = %q
  service_name = %q
  topic_name   = "%s-${count.index}"
  partitions   = 3
  replication  = 2
}`, projectName, kafkaName, topicName)
}

func testAccKafkaTopicResource(orgName, projectName, kafkaName, rName string) string {
	return fmt.Sprintf(`
data "aiven_organization" "foo" {
  name = %[1]q
}

data "aiven_kafka" "bar" {
  project      = %[2]q
  service_name = %[3]q
}

resource "aiven_organization_user_group" "foo" {
  name            = "test-acc-u-grp-%[4]s"
  organization_id = data.aiven_organization.foo.id
  description     = "test"
}

resource "aiven_kafka_topic" "foo" {
  project      = data.aiven_kafka.bar.project
  service_name = data.aiven_kafka.bar.service_name
  topic_name   = "test-acc-topic-%[4]s"
  partitions   = 3
  replication  = 2

  config {
    flush_ms                  = 10
    cleanup_policy            = "compact"
    min_cleanable_dirty_ratio = 0.01
    delete_retention_ms       = 50000
    retention_bytes           = 1234
    segment_bytes             = 1610612736
  }
}

data "aiven_kafka_topic" "topic" {
  project      = aiven_kafka_topic.foo.project
  service_name = aiven_kafka_topic.foo.service_name
  topic_name   = aiven_kafka_topic.foo.topic_name

  depends_on = [aiven_kafka_topic.foo]
}

resource "aiven_kafka_topic" "topic2" {
  project             = data.aiven_kafka.bar.project
  service_name        = data.aiven_kafka.bar.service_name
  topic_name          = "test-acc-topic2-%[4]s"
  topic_description   = "test-acc-topic2-desc-%[4]s"
  owner_user_group_id = aiven_organization_user_group.foo.group_id
  partitions          = 3
  replication         = 2
}`, orgName, projectName, kafkaName, rName)
}

func testAccKafkaTopicTerminationProtectionResource(projectName, kafkaName, topicName string) string {
	return fmt.Sprintf(`
resource "aiven_kafka_topic" "foo" {
  project                = %[1]q
  service_name           = %[2]q
  topic_name             = %[3]q
  partitions             = 3
  replication            = 2
  termination_protection = true
}

data "aiven_kafka_topic" "topic" {
  project      = aiven_kafka_topic.foo.project
  service_name = aiven_kafka_topic.foo.service_name
  topic_name   = aiven_kafka_topic.foo.topic_name

  depends_on = [aiven_kafka_topic.foo]
}`, projectName, kafkaName, topicName)
}

func testAccCheckAivenKafkaTopicAttributes(n string) resource.TestCheckFunc {
	return func(s *terraform.State) error {
		r := s.RootModule().Resources[n]
		a := r.Primary.Attributes

		log.Printf("[DEBUG] kafka topic attributes %v", a)

		if a["project"] == "" {
			return fmt.Errorf("expected to get a project name from Aiven")
		}

		if a["service_name"] == "" {
			return fmt.Errorf("expected to get a service_name from Aiven")
		}

		if a["topic_name"] == "" {
			return fmt.Errorf("expected to get a topic_name from Aiven")
		}

		if a["partitions"] == "" {
			return fmt.Errorf("expected to get a partitions from Aiven")
		}

		if a["replication"] == "" {
			return fmt.Errorf("expected to get a replication from Aiven")
		}

		return nil
	}
}

func testAccCheckAivenKafkaTopicResourceDestroy(s *terraform.State) error {
	c := acc.GetTestAivenClient()

	ctx := context.Background()

	// loop through the resources in state, verifying each created resource is destroyed
	for _, rs := range s.RootModule().Resources {
		if rs.Type == "aiven_kafka_topic" {
			project, serviceName, topicName, err := schemautil.SplitResourceID3(rs.Primary.ID)
			if err != nil {
				return err
			}

			_, err = c.Services.Get(ctx, project, serviceName)
			if err != nil {
				if aiven.IsNotFound(err) {
					return nil
				}
				return err
			}

			t, err := c.KafkaTopics.Get(ctx, project, serviceName, topicName)
			if err != nil {
				if aiven.IsNotFound(err) {
					return nil
				}
				return err
			}

			if t != nil {
				return fmt.Errorf("kafka topic (%s) still exists, id %s", topicName, rs.Primary.ID)
			}
		}
		if rs.Type == "aiven_organization_user_group" {
			orgID, userGroupID, err := schemautil.SplitResourceID2(rs.Primary.ID)
			if err != nil {
				return err
			}

			r, err := c.OrganizationUserGroups.Get(ctx, orgID, userGroupID)
			if err != nil {
				if aiven.IsNotFound(err) {
					return nil
				}
				return err
			}

			if r != nil {
				return fmt.Errorf("organization user group (%s) still exists", rs.Primary.ID)
			}
		}
	}

	return nil
}

func testAccAivenKafkaTopicResourceRecreateMissing(projectName, kafkaName, topicName string) string {
	return fmt.Sprintf(`
resource "aiven_kafka_topic" "topic" {
  project      = %[1]q
  service_name = %[2]q
  topic_name   = %[3]q
  partitions   = 5
  replication  = 2
}`, projectName, kafkaName, topicName)
}

func testAccAivenKafkaTopicResourceImportMissing(projectName, kafkaName, topicName string) string {
	result := fmt.Sprintf(`
resource "aiven_kafka_topic" "topic" {
  project      = %[1]q
  service_name = %[2]q
  topic_name   = %[3]q
  partitions   = 5
  replication  = 2
}
`, projectName, kafkaName, topicName)
	return result
}

func testAccAivenKafkaTopicConflictsIfExists(projectName, kafkaName, topicName string) string {
	return fmt.Sprintf(`
resource "aiven_kafka_topic" "topic" {
  project      = %[1]q
  service_name = %[2]q
  topic_name   = %[3]q
  partitions   = 5
  replication  = 2
}

resource "aiven_kafka_topic" "topic_conflict" {
  project      = %[1]q
  service_name = %[2]q
  topic_name   = %[3]q
  partitions   = 5
  replication  = 2

  depends_on = [
    aiven_kafka_topic.topic
  ]
}
`, projectName, kafkaName, topicName)
}

func TestAccAivenKafkaTopic_local_retention_bytes_overflow_error(t *testing.T) {
	project := acc.ProjectName()
	resource.ParallelTest(t, resource.TestCase{
		PreCheck:                 func() { acc.TestAccPreCheck(t) },
		ProtoV6ProviderFactories: acc.TestProtoV6ProviderFactories,
		CheckDestroy:             testAccCheckAivenKafkaTopicResourceDestroy,
		Steps: []resource.TestStep{
			{
				Config: fmt.Sprintf(`
resource "aiven_kafka_topic" "topic" {
  project      = %q
  service_name = "kafka-1c4a85e2"
  topic_name   = "foo"
  partitions   = 5
  replication  = 2

  config {
    local_retention_bytes = 3
    retention_bytes       = 1
  }

}`, project),
				ExpectError: regexp.MustCompile(`local_retention_bytes must not be more than retention_bytes value`),
			},
		},
	})
}

func TestAccAivenKafkaTopic_local_retention_bytes_overflow_dependency(t *testing.T) {
	project := acc.ProjectName()
	resource.ParallelTest(t, resource.TestCase{
		PreCheck:                 func() { acc.TestAccPreCheck(t) },
		ProtoV6ProviderFactories: acc.TestProtoV6ProviderFactories,
		CheckDestroy:             testAccCheckAivenKafkaTopicResourceDestroy,
		Steps: []resource.TestStep{
			{
				Config: fmt.Sprintf(`
resource "aiven_kafka_topic" "topic" {
  project      = %q
  service_name = "kafka-1c4a85e2"
  topic_name   = "foo"
  partitions   = 5
  replication  = 2

  config {
    local_retention_bytes = 3
  }

}`, project),
				ExpectError: regexp.MustCompile(`local_retention_bytes can't be set without retention_bytes`),
			},
		},
	})
}

func TestFlattenKafkaTopicConfig(t *testing.T) {
	cases := []struct {
		name   string
		expect map[string]any
		config aiven.KafkaTopicConfigResponse
	}{
		{
			name: "all fields",
			expect: map[string]any{
				"cleanup_policy":                      "foo",
				"compression_type":                    "bar",
				"delete_retention_ms":                 "0",
				"file_delete_delay_ms":                "1",
				"flush_messages":                      "2",
				"flush_ms":                            "3",
				"index_interval_bytes":                "4",
				"local_retention_bytes":               "5",
				"local_retention_ms":                  "6",
				"max_compaction_lag_ms":               "7",
				"max_message_bytes":                   "8",
				"message_downconversion_enable":       false,
				"message_format_version":              "",
				"message_timestamp_difference_max_ms": "0",
				"message_timestamp_type":              "",
				"min_cleanable_dirty_ratio":           0.2,
				"min_compaction_lag_ms":               "0",
				"min_insync_replicas":                 "0",
				"preallocate":                         true,
				"remote_storage_enable":               false,
				"retention_bytes":                     "0",
				"retention_ms":                        "0",
				"segment_bytes":                       "0",
				"segment_index_bytes":                 "0",
				"segment_jitter_ms":                   "0",
				"segment_ms":                          "0",
				"unclean_leader_election_enable":      true,
			},
			config: aiven.KafkaTopicConfigResponse{
				CleanupPolicy:                   &aiven.KafkaTopicConfigResponseString{Value: "foo"},
				CompressionType:                 &aiven.KafkaTopicConfigResponseString{Value: "bar"},
				DeleteRetentionMs:               &aiven.KafkaTopicConfigResponseInt{Value: 0},
				FileDeleteDelayMs:               &aiven.KafkaTopicConfigResponseInt{Value: 1},
				FlushMessages:                   &aiven.KafkaTopicConfigResponseInt{Value: 2},
				FlushMs:                         &aiven.KafkaTopicConfigResponseInt{Value: 3},
				IndexIntervalBytes:              &aiven.KafkaTopicConfigResponseInt{Value: 4},
				LocalRetentionBytes:             &aiven.KafkaTopicConfigResponseInt{Value: 5},
				LocalRetentionMs:                &aiven.KafkaTopicConfigResponseInt{Value: 6},
				MaxCompactionLagMs:              &aiven.KafkaTopicConfigResponseInt{Value: 7},
				MaxMessageBytes:                 &aiven.KafkaTopicConfigResponseInt{Value: 8},
				MessageDownconversionEnable:     &aiven.KafkaTopicConfigResponseBool{Value: false},
				MessageFormatVersion:            &aiven.KafkaTopicConfigResponseString{Value: ""},
				MessageTimestampDifferenceMaxMs: &aiven.KafkaTopicConfigResponseInt{Value: 0},
				MessageTimestampType:            &aiven.KafkaTopicConfigResponseString{Value: ""},
				MinCleanableDirtyRatio:          &aiven.KafkaTopicConfigResponseFloat{Value: 0.2},
				MinCompactionLagMs:              &aiven.KafkaTopicConfigResponseInt{Value: 0},
				MinInsyncReplicas:               &aiven.KafkaTopicConfigResponseInt{Value: 0},
				Preallocate:                     &aiven.KafkaTopicConfigResponseBool{Value: true},
				RemoteStorageEnable:             &aiven.KafkaTopicConfigResponseBool{Value: false},
				RetentionBytes:                  &aiven.KafkaTopicConfigResponseInt{Value: 0},
				RetentionMs:                     &aiven.KafkaTopicConfigResponseInt{Value: 0},
				SegmentBytes:                    &aiven.KafkaTopicConfigResponseInt{Value: 0},
				SegmentIndexBytes:               &aiven.KafkaTopicConfigResponseInt{Value: 0},
				SegmentJitterMs:                 &aiven.KafkaTopicConfigResponseInt{Value: 0},
				SegmentMs:                       &aiven.KafkaTopicConfigResponseInt{Value: 0},
				UncleanLeaderElectionEnable:     &aiven.KafkaTopicConfigResponseBool{Value: true},
			},
		},
		{
			name: "few fields",
			expect: map[string]any{
				"local_retention_bytes": "1",
				"retention_bytes":       "2",
				"segment_bytes":         "10",
			},
			config: aiven.KafkaTopicConfigResponse{
				LocalRetentionBytes: &aiven.KafkaTopicConfigResponseInt{Value: 1},
				RetentionBytes:      &aiven.KafkaTopicConfigResponseInt{Value: 2},
				SegmentBytes:        &aiven.KafkaTopicConfigResponseInt{Value: 10},
			},
		},
	}

	for _, opt := range cases {
		t.Run(opt.name, func(t *testing.T) {
			result, err := kafkatopic.FlattenKafkaTopicConfig(&aiven.KafkaTopic{Config: opt.config})
			require.NoError(t, err)
			assert.Empty(t, cmp.Diff([]map[string]any{opt.expect}, result))
		})
	}
}
