refactoring

Author: rasha108bik

internal/resources/topic/consumer/create.go

@@ -0,0 +1,74 @@
+package consumer
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/ydb-platform/terraform-provider-ydb/internal/helpers/topic"
+	tbl "github.com/ydb-platform/terraform-provider-ydb/internal/table"
+	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions"
+	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topictypes"
+)
+
+const (
+	ydbTopicDefaultMaxPartitionWriteSpeed = 1048576
+)
+
+func (h *handler) Create(ctx context.Context, d *schema.ResourceData, _ interface{}) diag.Diagnostics {
+	tableResource, err := tableResourceSchemaToTableResource(d)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	if tableResource == nil {
+		return diag.Diagnostics{
+			diag.Diagnostic{
+				Severity: diag.Error,
+				Summary:  "got nil resource, unreachable code",
+			},
+		}
+	}
+	db, err := tbl.CreateDBConnection(ctx, tbl.ClientParams{
+		DatabaseEndpoint: tableResource.DatabaseEndpoint,
+		Token:            h.token,
+	})
+	if err != nil {
+		return diag.Diagnostics{
+			diag.Diagnostic{
+				Severity: diag.Error,
+				Summary:  "failed to initialize table client",
+				Detail:   err.Error(),
+			},
+		}
+	}
+	defer func() {
+		_ = db.Close(ctx)
+	}()
+
+	var supportedCodecs []topictypes.Codec
+	if gotCodecs, ok := d.GetOk("supported_codecs"); !ok {
+		supportedCodecs = topic.YDBTopicDefaultCodecs
+	} else {
+		for _, c := range gotCodecs.([]interface{}) {
+			cod := c.(string)
+			supportedCodecs = append(supportedCodecs, topic.YDBTopicCodecNameToCodec[cod])
+		}
+	}
+
+	err = db.Topic().Create(ctx, d.Get("name").(string),
+		topicoptions.CreateWithSupportedCodecs(supportedCodecs...),
+		topicoptions.CreateWithPartitionWriteBurstBytes(ydbTopicDefaultMaxPartitionWriteSpeed),
+		topicoptions.CreateWithPartitionWriteSpeedBytesPerSecond(ydbTopicDefaultMaxPartitionWriteSpeed),
+		topicoptions.CreateWithRetentionPeriod(time.Duration(d.Get("starting_message_timestamp_ms").(int))*time.Millisecond),
+	)
+	if err != nil {
+		return diag.FromErr(fmt.Errorf("failed to initialize ydb-topic-consumer control plane client: %w", err))
+	}
+
+	topicPath := d.Get("topic_path").(string)
+	d.SetId(d.Get("connection_string").(string) + "&path=" + topicPath)
+
+	return h.Read(ctx, d, nil)
+}

internal/resources/topic/consumer/delete.go

@@ -0,0 +1,44 @@
+package consumer
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	tbl "github.com/ydb-platform/terraform-provider-ydb/internal/table"
+)
+
+func (h *handler) Delete(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	tableResource, err := tableResourceSchemaToTableResource(d)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	if tableResource == nil {
+		return diag.Diagnostics{
+			diag.Diagnostic{
+				Severity: diag.Error,
+				Summary:  "got nil resource, unreachable code",
+			},
+		}
+	}
+
+	db, err := tbl.CreateDBConnection(ctx, tbl.ClientParams{
+		DatabaseEndpoint: tableResource.DatabaseEndpoint,
+		Token:            h.token,
+	})
+	if err != nil {
+		return diag.Errorf("failed to initialize table client: %s", err)
+	}
+	defer func() {
+		_ = db.Close(ctx)
+	}()
+
+	topicName := tableResource.Entity.GetEntityPath()
+	err = db.Topic().Drop(ctx, topicName)
+	if err != nil {
+		return diag.FromErr(fmt.Errorf("failed to delete topic: %w", err))
+	}
+
+	return nil
+}

sdk/terraform/topic/consumer_structures_manual.go renamed to internal/resources/topic/consumer/handler.go

@@ -1,4 +1,4 @@
-package topic
+package consumer
 
 import (
 	"fmt"
@@ -7,12 +7,30 @@ import (
 
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/ydb-platform/terraform-provider-ydb/internal/helpers/topic"
+	"github.com/ydb-platform/terraform-provider-ydb/internal/resources"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topictypes"
 )
 
+type handler struct {
+	token string
+}
+
+func NewHandler(token string) resources.Handler {
+	return &handler{
+		token: token,
+	}
+}
+
+const (
+	ydbTopicCodecGZIP = "gzip"
+	ydbTopicCodecRAW  = "raw"
+	ydbTopicCodecZSTD = "zstd"
+)
+
 func flattenYDBTopicConsumerDescription(d *schema.ResourceData, desc topictypes.TopicDescription) error {
 	_ = d.Set("name", d.Get("name").(string)) // NOTE(shmel1k@): TopicService SDK does not return path for stream.
+	_ = d.Set("topic_path", d.Get("topic_path").(string))
 
 	supportedCodecs := make([]string, 0, len(desc.SupportedCodecs))
 	for _, v := range desc.SupportedCodecs {
@@ -31,7 +49,7 @@ func flattenYDBTopicConsumerDescription(d *schema.ResourceData, desc topictypes.
 		return err
 	}
 
-	return d.Set("database_endpoint", d.Get("database_endpoint").(string))
+	return d.Set("connection_string", d.Get("connection_string").(string))
 }
 
 func prepareYDBTopicConsumerAlterSettings(

internal/resources/topic/consumer/read.go

@@ -0,0 +1,55 @@
+package consumer
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+
+	tbl "github.com/ydb-platform/terraform-provider-ydb/internal/table"
+)
+
+func (h *handler) Read(ctx context.Context, d *schema.ResourceData, _ interface{}) diag.Diagnostics {
+	tableResource, err := tableResourceSchemaToTableResource(d)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	if tableResource == nil {
+		return diag.Diagnostics{
+			diag.Diagnostic{
+				Severity: diag.Error,
+				Summary:  "got nil resource, unreachable code",
+			},
+		}
+	}
+
+	db, err := tbl.CreateDBConnection(ctx, tbl.ClientParams{
+		DatabaseEndpoint: tableResource.DatabaseEndpoint,
+		Token:            h.token,
+	})
+	if err != nil {
+		return diag.Errorf("failed to initialize table client: %s", err)
+	}
+	defer func() {
+		_ = db.Close(ctx)
+	}()
+
+	topicName := tableResource.Entity.GetEntityPath()
+	description, err := db.Topic().Describe(ctx, topicName)
+	if err != nil {
+		if strings.Contains(err.Error(), "does not exist") {
+			d.SetId("") // marking as non-existing resource.
+			return nil
+		}
+		return diag.FromErr(fmt.Errorf("resource: failed to describe topic: %w", err))
+	}
+
+	err = flattenYDBTopicConsumerDescription(d, description)
+	if err != nil {
+		return diag.FromErr(fmt.Errorf("failed to flatten topic description: %w", err))
+	}
+
+	return nil
+}

internal/resources/topic/consumer/table.go

@@ -0,0 +1,40 @@
+package consumer
+
+import (
+	"fmt"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/ydb-platform/terraform-provider-ydb/internal/helpers"
+)
+
+type Resource struct {
+	Entity *helpers.YDBEntity
+
+	FullPath         string
+	Path             string
+	DatabaseEndpoint string
+}
+
+func tableResourceSchemaToTableResource(d *schema.ResourceData) (*Resource, error) {
+	var entity *helpers.YDBEntity
+	var err error
+	if d.Id() != "" {
+		entity, err = helpers.ParseYDBEntityID(d.Id())
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse table entity: %w", err)
+		}
+	}
+
+	var databaseEndpoint string
+	if entity != nil {
+		databaseEndpoint = entity.PrepareFullYDBEndpoint()
+	} else {
+		// NOTE(shmel1k@): resource is not initialized yet.
+		databaseEndpoint = d.Get("connection_string").(string)
+	}
+
+	return &Resource{
+		Entity:           entity,
+		DatabaseEndpoint: databaseEndpoint,
+	}, nil
+}

internal/resources/topic/consumer/update.go

@@ -0,0 +1,64 @@
+package consumer
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	tbl "github.com/ydb-platform/terraform-provider-ydb/internal/table"
+)
+
+func (h *handler) performYDBTopicConsumerUpdate(ctx context.Context, d *schema.ResourceData) diag.Diagnostics {
+	tableResource, err := tableResourceSchemaToTableResource(d)
+	if err != nil {
+		return diag.FromErr(err)
+	}
+	if tableResource == nil {
+		return diag.Diagnostics{
+			diag.Diagnostic{
+				Severity: diag.Error,
+				Summary:  "got nil resource, unreachable code",
+			},
+		}
+	}
+
+	db, err := tbl.CreateDBConnection(ctx, tbl.ClientParams{
+		DatabaseEndpoint: tableResource.DatabaseEndpoint,
+		Token:            h.token,
+	})
+	if err != nil {
+		return diag.Errorf("failed to initialize table client: %s", err)
+	}
+	defer func() {
+		_ = db.Close(ctx)
+	}()
+
+	if d.HasChange("name") {
+		// Creating new topic
+		return h.Create(ctx, d, nil)
+	}
+
+	topicName := tableResource.Entity.GetEntityPath()
+	desc, err := db.Topic().Describe(ctx, topicName)
+	if err != nil {
+		if strings.Contains(err.Error(), "does not exist") {
+			return h.Create(ctx, d, nil)
+		}
+		return diag.FromErr(fmt.Errorf("failed to get description for topic %q", topicName))
+	}
+
+	opts := prepareYDBTopicConsumerAlterSettings(d, desc)
+	err = db.Topic().Alter(ctx, topicName, opts...)
+	if err != nil {
+		return diag.FromErr(fmt.Errorf("got error when tried to alter topic: %w", err))
+	}
+
+	return h.Read(ctx, d, nil)
+}
+
+func (h *handler) Update(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	_ = meta
+	return h.performYDBTopicConsumerUpdate(ctx, d)
+}