Adding support for creating queue consumers

.changelog/3833.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/cloudflare_queue: added support for create queue consumers
+```

.gitignore

@@ -35,3 +35,5 @@ dist/*
 .envrc
 .flox/
 .vscode/private.env
+
+setvars.sh

internal/sdkv2provider/resource_cloudflare_queue.go

@@ -31,7 +31,7 @@ func resourceCloudflareQueueCreate(ctx context.Context, d *schema.ResourceData,
 	client := meta.(*cloudflare.API)
 	accountID := d.Get(consts.AccountIDSchemaKey).(string)
 	queueName := d.Get("name").(string)
-
+	
 	req := cloudflare.CreateQueueParams{
 		Name: queueName,
 	}
@@ -46,6 +46,35 @@ func resourceCloudflareQueueCreate(ctx context.Context, d *schema.ResourceData,
 		return diag.FromErr(fmt.Errorf("failed to find id in Create response; resource was empty"))
 	}
 
+	if d.Get("consumer") != nil {
+		consumers := d.Get("consumer").([]interface{})
+		for _, consumer := range consumers {
+			consumerMap := consumer.(map[string]interface{})
+			settings := consumerMap["settings"].([]interface{})[0].(map[string]interface{})
+			req := cloudflare.CreateQueueConsumerParams{
+				QueueName: queueName,
+				Consumer: cloudflare.QueueConsumer{
+					Service: "workers",
+					Name: consumerMap["script_name"].(string),
+					QueueName: queueName,
+					Environment: consumerMap["environment"].(string),
+					ScriptName: consumerMap["script_name"].(string),
+					Settings: cloudflare.QueueConsumerSettings{
+						BatchSize: settings["batch_size"].(int),
+						MaxRetires: settings["max_retries"].(int),
+						MaxWaitTime: settings["max_wait_time_ms"].(int),
+					},
+				},
+			}
+
+			_, err := client.CreateQueueConsumer(ctx, cloudflare.AccountIdentifier(accountID), req)
+			// println(fmt.Sprintf("result: %+v", result))
+			if err != nil {
+				return diag.FromErr(errors.Wrap(err, "error creating workers queue consumer"))
+			}
+		}
+	}
+
 	d.SetId(r.ID)
 
 	tflog.Info(ctx, fmt.Sprintf("Cloudflare Workers Queue ID: %s. Name: %s", d.Id(), queueName))
@@ -62,13 +91,32 @@ func resourceCloudflareQueueRead(ctx context.Context, d *schema.ResourceData, me
 	if err != nil {
 		return diag.FromErr(errors.Wrap(err, "error reading queues"))
 	}
-
+	
 	var queue cloudflare.Queue
 	for _, r := range resp {
 		if r.ID == queueID {
 			queue = r
 			d.Set("name", r.Name)
-			break
+			consumers := make([]map[string]interface{}, 0)
+			if len(r.Consumers) > 0 {
+				for _, consumer := range r.Consumers {
+					consumerMap := map[string]interface{}{
+						"created_on": consumer.CreatedOn,
+						"environment": consumer.Environment,
+						"queue_name": consumer.QueueName,
+						"script_name": consumer.ScriptName,
+						"settings": []map[string]interface{}{
+							{
+								"batch_size": consumer.Settings.BatchSize,
+								"max_retries": consumer.Settings.MaxRetires,
+								"max_wait_time_ms": consumer.Settings.MaxWaitTime,
+							},
+						},
+					}
+					consumers = append(consumers, consumerMap)
+				}
+				d.Set("consumer", consumers)
+			}
 		}
 	}
 

internal/sdkv2provider/resource_cloudflare_queue_test.go

@@ -80,6 +80,37 @@ func TestAccCloudflareQueue_Basic(t *testing.T) {
 	})
 }
 
+func TestAccCloudflareQueue_Consumer(t *testing.T) {
+	t.Parallel()
+	var queue cloudflare.Queue
+	accountID := os.Getenv("CLOUDFLARE_ACCOUNT_ID")
+	rnd := generateRandomResourceName()
+	resourceName := "cloudflare_queue." + rnd
+
+	resource.Test(t, resource.TestCase{
+		PreCheck:          func() { testAccPreCheck(t) },
+		ProviderFactories: providerFactories,
+		CheckDestroy:      testAccCloudflareQueueDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccCheckCloudflareQueueWConsumer(rnd, accountID, rnd),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckCloudflareQueueExists(rnd, &queue),
+					resource.TestCheckResourceAttr(resourceName, "name", rnd),
+					resource.TestCheckResourceAttr(resourceName, "account_id", accountID),
+				),
+			},
+
+			{
+				ImportState:         true,
+				ImportStateVerify:   true,
+				ResourceName:        resourceName,
+				ImportStateIdPrefix: fmt.Sprintf("%s/", accountID),
+			},
+		},
+	})
+}
+
 func testAccCloudflareQueueDestroy(s *terraform.State) error {
 	client := testAccProvider.Meta().(*cloudflare.API)
 
@@ -112,6 +143,34 @@ resource "cloudflare_queue" "%[1]s" {
 }`, rnd, accountID, name)
 }
 
+func testAccCheckCloudflareQueueWConsumer(rnd, accountID, name string) string {
+	queueModuleContent := `export default { queue(batch, env) { return new Response('Hello world'); }, };`
+	tf_content := fmt.Sprintf(`
+	resource "cloudflare_workers_script" "%[1]s" {
+		  account_id = "%[2]s"
+		  name = "%[3]s"
+		  content = "%[4]s"
+		  module = true
+		  compatibility_date = "2023-03-19"
+		  compatibility_flags = ["nodejs_compat", "web_socket_compression"]
+	}
+
+	resource "cloudflare_queue" "%[1]s" {
+		account_id = "%[2]s"
+		name = "%[3]s"
+		consumer {
+			script_name = "%[1]s"
+			settings {
+				batch_size = 1
+				max_retries = 1
+				max_wait_time_ms = 1000
+			}
+		}
+		depends_on = [cloudflare_workers_script.%[1]s]
+	}`, rnd, accountID, name, queueModuleContent)
+	return tf_content
+}
+
 func testAccCheckCloudflareQueueExists(name string, queue *cloudflare.Queue) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
 		client := testAccProvider.Meta().(*cloudflare.API)

internal/sdkv2provider/schema_cloudflare_queue.go

@@ -17,5 +17,68 @@ func resourceCloudflareQueueSchema() map[string]*schema.Schema {
 			Required:    true,
 			Description: "The name of the queue.",
 		},
+		"consumer": {
+			Type:        schema.TypeList,
+			Optional:    true,
+			Description: "Array of consumers.",
+			Elem: &schema.Resource{
+				Schema: map[string]*schema.Schema{
+					"created_on": {
+						Description: "Created on date",
+						Type:        schema.TypeString,
+						Optional:    true,
+						ForceNew:    true,
+					},
+					"environment": {
+						Description: "Environment",
+						Type:        schema.TypeString,
+						Optional:    true,
+						ForceNew:    true,
+					},
+					"queue_name": {
+						Description: "Queue name",
+						Type:        schema.TypeString,
+						Optional:    true,
+						ForceNew:    true,
+					},
+					"script_name": {
+						Description: "script_name",
+						Type:        schema.TypeString,
+						Optional:    true,
+						ForceNew:    true,
+					},
+					"settings": {
+						Description: "Settings",
+						Type:        schema.TypeList,
+						Optional:    true,
+						ForceNew:    true,
+						Elem: &schema.Resource{
+							Schema: map[string]*schema.Schema{
+								"batch_size": {
+									Description: "Batch size",
+									Type:        schema.TypeInt,
+									Optional:    true,
+									ForceNew:    true,
+								},
+								"max_retries": {
+									Description: "Max retries",
+									Type:        schema.TypeInt,
+									Optional:    true,
+									ForceNew:    true,
+								},
+								"max_wait_time_ms": {
+									Description: "Max wait time in milliseconds",
+									Type:        schema.TypeInt,
+									Optional:    true,
+									ForceNew:    true,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
 	}
 }
+
+