Add support for Cloudflare Queues

.changelog/2134.txt

@@ -0,0 +1,3 @@
+```release-note:new-resource
+cloudflare_workers_queue
+```

internal/provider/provider.go

@@ -279,6 +279,7 @@ func New(version string) func() *schema.Provider {
 				"cloudflare_worker_script":                          resourceCloudflareWorkerScript(),
 				"cloudflare_workers_kv_namespace":                   resourceCloudflareWorkersKVNamespace(),
 				"cloudflare_workers_kv":                             resourceCloudflareWorkerKV(),
+				"cloudflare_workers_queue":                          resourceCloudflareWorkersQueue(),
 				"cloudflare_zone_cache_variants":                    resourceCloudflareZoneCacheVariants(),
 				"cloudflare_zone_dnssec":                            resourceCloudflareZoneDNSSEC(),
 				"cloudflare_zone_lockdown":                          resourceCloudflareZoneLockdown(),

internal/provider/resource_cloudflare_workers_queue.go

@@ -0,0 +1,131 @@
+package provider
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cloudflare/cloudflare-go"
+	"github.com/hashicorp/terraform-plugin-log/tflog"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/pkg/errors"
+)
+
+func resourceCloudflareWorkersQueue() *schema.Resource {
+	return &schema.Resource{
+		Schema:        resourceCloudflareWorkersQueueSchema(),
+		CreateContext: resourceCloudflareWorkersQueueCreate,
+		ReadContext:   resourceCloudflareWorkersQueueRead,
+		UpdateContext: resourceCloudflareWorkersQueueUpdate,
+		DeleteContext: resourceCloudflareWorkersQueueDelete,
+		Importer: &schema.ResourceImporter{
+			StateContext: resourceCloudflareWorkersQueueImport,
+		},
+		Description: "Provides the ability to manage Cloudflare Workers Queue features.",
+	}
+}
+
+func resourceCloudflareWorkersQueueCreate(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	client := meta.(*cloudflare.API)
+
+	accountID := d.Get("account_id").(string)
+	if accountID == "" {
+		accountID = client.AccountID
+	}
+
+	queueName := d.Get("name").(string)
+	req := cloudflare.CreateQueueParams{
+		Name: queueName,
+	}
+
+	tflog.Debug(ctx, fmt.Sprintf("[Info] Creating Cloudflare Workers Queue from struct: %+v", req))
+
+	r, err := client.CreateQueue(ctx, cloudflare.AccountIdentifier(accountID), req)
+	if err != nil {
+		return diag.FromErr(errors.Wrap(err, "error creating workers queue"))
+	}
+
+	if r.ID == "" {
+		return diag.FromErr(fmt.Errorf("failed to find id in Create response; resource was empty"))
+	}
+
+	d.SetId(r.ID)
+
+	tflog.Info(ctx, fmt.Sprintf("Cloudflare Workers Queue ID: %s. Name: %s", d.Id(), queueName))
+
+	return resourceCloudflareWorkersQueueRead(ctx, d, meta)
+}
+
+func resourceCloudflareWorkersQueueRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	client := meta.(*cloudflare.API)
+	queueID := d.Id()
+
+	accountID := d.Get("account_id").(string)
+	if accountID == "" {
+		accountID = client.AccountID
+	}
+
+	resp, _, err := client.ListQueues(ctx, cloudflare.AccountIdentifier(accountID), cloudflare.ListQueuesParams{})
+	if err != nil {
+		return diag.FromErr(errors.Wrap(err, "error reading workers "))
+	}
+
+	var queue cloudflare.Queue
+	for _, r := range resp {
+		if r.ID == queueID {
+			queue = r
+			break
+		}
+	}
+
+	if queue.ID == "" {
+		d.SetId("")
+		return nil
+	}
+
+	d.Set("account_id", accountID)
+
+	return nil
+}
+
+func resourceCloudflareWorkersQueueUpdate(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	client := meta.(*cloudflare.API)
+
+	accountID := d.Get("account_id").(string)
+	if accountID == "" {
+		accountID = client.AccountID
+	}
+
+	// TODO(soon) fix the cloudflare-go UpdateQueue implementation: updating a queue should accept the existing name, as well as the new name. Other properties are read-only.
+	_, err := client.UpdateQueue(ctx, cloudflare.AccountIdentifier(accountID), cloudflare.UpdateQueueParams{
+		Name: d.Get("name").(string),
+	})
+	if err != nil {
+		return diag.FromErr(errors.Wrap(err, "error updating workers queue"))
+	}
+
+	return resourceCloudflareWorkersQueueRead(ctx, d, meta)
+}
+
+func resourceCloudflareWorkersQueueDelete(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	client := meta.(*cloudflare.API)
+	accountID := d.Get("account_id").(string)
+	if accountID == "" {
+		accountID = client.AccountID
+	}
+
+	tflog.Info(ctx, fmt.Sprintf("Deleting Cloudflare Workers Queue with id: %+v", d.Id()))
+
+	err := client.DeleteQueue(ctx, cloudflare.AccountIdentifier(accountID), d.Get("name").(string))
+	if err != nil {
+		return diag.FromErr(errors.Wrap(err, "error deleting workers queue"))
+	}
+
+	d.SetId("")
+	return nil
+}
+
+// TODO(now) is this needed?
+func resourceCloudflareWorkersQueueImport(ctx context.Context, d *schema.ResourceData, meta interface{}) ([]*schema.ResourceData, error) {
+	return []*schema.ResourceData{d}, nil
+}

internal/provider/resource_cloudflare_workers_queue_test.go

@@ -0,0 +1,95 @@
+package provider
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/cloudflare/cloudflare-go"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
+)
+
+func TestAccCloudflareWorkersQueue_Basic(t *testing.T) {
+	t.Parallel()
+	var queue cloudflare.Queue
+	rnd := generateRandomResourceName()
+	resourceName := "cloudflare_workers_queue." + rnd
+	resource.Test(t, resource.TestCase{
+		PreCheck:          func() { testAccPreCheck(t) },
+		ProviderFactories: providerFactories,
+		CheckDestroy:      testAccCloudflareWorkersQueueDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccCheckCloudflareWorkersQueue(rnd),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckCloudflareWorkersQueueExists(rnd, &queue),
+					resource.TestCheckResourceAttr(resourceName, "name", rnd),
+				),
+			},
+		},
+	})
+}
+
+func testAccCloudflareWorkersQueueDestroy(s *terraform.State) error {
+	client := testAccProvider.Meta().(*cloudflare.API)
+
+	for _, rs := range s.RootModule().Resources {
+		if rs.Type != "cloudflare_workers_queue" {
+			continue
+		}
+
+		accountID := rs.Primary.Attributes["account_id"]
+		if accountID == "" {
+			accountID = client.AccountID
+		}
+
+		resp, _, err := client.ListQueues(context.Background(), cloudflare.AccountIdentifier(accountID), cloudflare.ListQueuesParams{})
+		if err != nil {
+			return err
+		}
+
+		for _, n := range resp {
+			if n.ID == rs.Primary.ID {
+				return fmt.Errorf("queue still exists but should not")
+			}
+		}
+	}
+
+	return nil
+}
+
+func testAccCheckCloudflareWorkersQueue(rName string) string {
+	return fmt.Sprintf(`
+resource "cloudflare_workers_queue" "%[1]s" {
+	title = "%[1]s"
+}`, rName)
+}
+
+func testAccCheckCloudflareWorkersQueueExists(name string, queue *cloudflare.Queue) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		client := testAccProvider.Meta().(*cloudflare.API)
+
+		rs, ok := s.RootModule().Resources["cloudflare_workers_queue."+name]
+		if !ok {
+			return fmt.Errorf("not found: %s", name)
+		}
+		accountID := rs.Primary.Attributes["account_id"]
+		if accountID == "" {
+			accountID = client.AccountID
+		}
+		resp, _, err := client.ListQueues(context.Background(), cloudflare.AccountIdentifier(accountID), cloudflare.ListQueuesParams{})
+		if err != nil {
+			return err
+		}
+
+		for _, q := range resp {
+			if q.Name == name {
+				*queue = q
+				return nil
+			}
+		}
+
+		return fmt.Errorf("queue not found")
+	}
+}

internal/provider/resource_cloudflare_workers_script.go

@@ -119,6 +119,15 @@ func parseWorkerBindings(d *schema.ResourceData, bindings ScriptBindings) {
 			Dataset: data["dataset"].(string),
 		}
 	}
+
+	for _, rawData := range d.Get("queue_binding").(*schema.Set).List() {
+		data := rawData.(map[string]interface{})
+
+		bindings[data["binding"].(string)] = cloudflare.WorkerQueueBinding{
+			Binding: data["binding"].(string),
+			Queue:   data["queue"].(string),
+		}
+	}
 }
 
 func resourceCloudflareWorkerScriptCreate(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
@@ -209,6 +218,7 @@ func resourceCloudflareWorkerScriptRead(ctx context.Context, d *schema.ResourceD
 	serviceBindings := &schema.Set{F: schema.HashResource(serviceBindingResource)}
 	r2BucketBindings := &schema.Set{F: schema.HashResource(r2BucketBindingResource)}
 	analyticsEngineBindings := &schema.Set{F: schema.HashResource(analyticsEngineBindingResource)}
+	queueBindings := &schema.Set{F: schema.HashResource(queueBindingResource)}
 
 	for name, binding := range bindings {
 		switch v := binding.(type) {
@@ -257,6 +267,11 @@ func resourceCloudflareWorkerScriptRead(ctx context.Context, d *schema.ResourceD
 				"name":    name,
 				"dataset": v.Dataset,
 			})
+		case cloudflare.WorkerQueueBinding:
+			queueBindings.Add(map[string]interface{}{
+				"binding": name,
+				"queue":   v.Queue,
+			})
 		}
 	}
 
@@ -292,6 +307,10 @@ func resourceCloudflareWorkerScriptRead(ctx context.Context, d *schema.ResourceD
 		return diag.FromErr(fmt.Errorf("cannot set analytics engine bindings (%s): %w", d.Id(), err))
 	}
 
+	if err := d.Set("queue_binding", queueBindings); err != nil {
+		return diag.FromErr(fmt.Errorf("cannot set queue bindings (%s): %w", d.Id(), err))
+	}
+
 	d.SetId(scriptData.ID)
 
 	return nil

internal/provider/resource_cloudflare_workers_script_test.go

@@ -55,7 +55,7 @@ func TestAccCloudflareWorkerScript_MultiScriptEnt(t *testing.T) {
 			{
 				Config: testAccCheckCloudflareWorkerScriptConfigMultiScriptUpdateBinding(rnd, accountID),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckCloudflareWorkerScriptExists(name, &script, []string{"MY_KV_NAMESPACE", "MY_PLAIN_TEXT", "MY_SECRET_TEXT", "MY_WASM", "MY_SERVICE_BINDING", "MY_BUCKET"}),
+					testAccCheckCloudflareWorkerScriptExists(name, &script, []string{"MY_KV_NAMESPACE", "MY_PLAIN_TEXT", "MY_SECRET_TEXT", "MY_WASM", "MY_SERVICE_BINDING", "MY_BUCKET", "MY_QUEUE"}),
 					resource.TestCheckResourceAttr(name, "name", rnd),
 					resource.TestCheckResourceAttr(name, "content", scriptContent2),
 				),
@@ -132,6 +132,11 @@ resource "cloudflare_workers_kv_namespace" "%[1]s" {
 	title = "%[1]s"
 }
 
+resource "cloudflare_workers_queue" "%[1]s" {
+	account_id = "%[4]s"
+	name = "%[1]s"
+}
+
 resource "cloudflare_worker_script" "%[1]s-service" {
 	account_id = "%[4]s"
 	name    = "%[1]s-service"
@@ -173,6 +178,12 @@ resource "cloudflare_worker_script" "%[1]s" {
     service = cloudflare_worker_script.%[1]s-service.name
     environment = "production"
   }
+
+  queue_binding {
+    binding         = "MY_QUEUE"
+    queue = cloudflare_workers_queue.%[1]s.name
+  }
+
 }`, rnd, scriptContent2, encodedWasm, accountID)
 }
 

internal/provider/schema_cloudflare_workers_queue.go

@@ -0,0 +1,19 @@
+package provider
+
+import "github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+
+func resourceCloudflareWorkersQueueSchema() map[string]*schema.Schema {
+	return map[string]*schema.Schema{
+		"account_id": {
+			Description: "The account identifier to target for the resource.",
+			Type:        schema.TypeString,
+			Optional:    true,
+			Computed:    true,
+		},
+		"name": {
+			Type:        schema.TypeString,
+			Required:    true,
+			Description: "The name of the queue.",
+		},
+	}
+}

internal/provider/schema_cloudflare_workers_script.go

@@ -113,6 +113,21 @@ var analyticsEngineBindingResource = &schema.Resource{
 	},
 }
 
+var queueBindingResource = &schema.Resource{
+	Schema: map[string]*schema.Schema{
+		"binding": {
+			Type:        schema.TypeString,
+			Required:    true,
+			Description: "The name of the global variable for the binding in your Worker code.",
+		},
+		"queue": {
+			Type:        schema.TypeString,
+			Required:    true,
+			Description: "Name of the queue you want to use.",
+		},
+	},
+}
+
 func resourceCloudflareWorkerScriptSchema() map[string]*schema.Schema {
 	return map[string]*schema.Schema{
 		"account_id": {
@@ -171,5 +186,10 @@ func resourceCloudflareWorkerScriptSchema() map[string]*schema.Schema {
 			Optional: true,
 			Elem:     analyticsEngineBindingResource,
 		},
+		"queue_binding": {
+			Type:     schema.TypeSet,
+			Optional: true,
+			Elem:     queueBindingResource,
+		},
 	}
 }