Initial webhook source setup

docs/resources/source_webhook.md

@@ -0,0 +1,100 @@
+---
+# generated by https://github.com/hashicorp/terraform-plugin-docs
+page_title: "materialize_source_webhook Resource - terraform-provider-materialize"
+subcategory: ""
+description: |-
+  A webhook source describes a webhook you want Materialize to read data from.
+---
+
+# materialize_source_webhook (Resource)
+
+A webhook source describes a webhook you want Materialize to read data from.
+
+## Example Usage
+
+```terraform
+resource "materialize_source_webhook" "example_webhook" {
+  name            = "example_webhook"
+  cluster_name    = materialize_cluster.example_cluster.name
+  body_format     = "json"
+  include_headers = false
+}
+```
+
+<!-- schema generated by tfplugindocs -->
+## Schema
+
+### Required
+
+- `body_format` (String) The body format of the webhook.
+- `name` (String) The identifier for the source.
+
+### Optional
+
+- `check_expression` (String) The check expression for the webhook.
+- `check_options` (Block List) The check options for the webhook. (see [below for nested schema](#nestedblock--check_options))
+- `cluster_name` (String) The cluster to maintain this source.
+- `database_name` (String) The identifier for the source database. Defaults to `MZ_DATABASE` environment variable if set or `materialize` if environment variable is not set.
+- `include_headers` (Boolean) Include headers in the webhook.
+- `ownership_role` (String) The owernship role of the object.
+- `schema_name` (String) The identifier for the source schema. Defaults to `public`.
+
+### Read-Only
+
+- `id` (String) The ID of this resource.
+- `qualified_sql_name` (String) The fully qualified name of the source.
+- `size` (String) The size of the source.
+- `subsource` (List of Object) Subsources of a source. (see [below for nested schema](#nestedatt--subsource))
+
+<a id="nestedblock--check_options"></a>
+### Nested Schema for `check_options`
+
+Required:
+
+- `field` (Block List, Min: 1, Max: 1) The field for the check options. (see [below for nested schema](#nestedblock--check_options--field))
+
+Optional:
+
+- `alias` (String) The alias for the check options.
+
+<a id="nestedblock--check_options--field"></a>
+### Nested Schema for `check_options.field`
+
+Optional:
+
+- `body` (Boolean) The body for the check options.
+- `headers` (Boolean) The headers for the check options.
+- `secret` (Block List, Max: 1) The secret for the check options. (see [below for nested schema](#nestedblock--check_options--field--secret))
+
+<a id="nestedblock--check_options--field--secret"></a>
+### Nested Schema for `check_options.field.secret`
+
+Required:
+
+- `name` (String) The secret name.
+
+Optional:
+
+- `database_name` (String) The secret database name.
+- `schema_name` (String) The secret schema name.
+
+
+
+
+<a id="nestedatt--subsource"></a>
+### Nested Schema for `subsource`
+
+Read-Only:
+
+- `database_name` (String)
+- `name` (String)
+- `schema_name` (String)
+
+## Import
+
+Import is supported using the following syntax:
+
+```shell
+# Sources can be imported using the source id:
+terraform import materialize_source_webhook.example_source_webhook <source_id>
+```

examples/resources/materialize_source_webhook/import.sh

@@ -0,0 +1,2 @@
+# Sources can be imported using the source id:
+terraform import materialize_source_webhook.example_source_webhook <source_id>

examples/resources/materialize_source_webhook/resource.tf

@@ -0,0 +1,6 @@
+resource "materialize_source_webhook" "example_webhook" {
+  name            = "example_webhook"
+  cluster_name    = materialize_cluster.example_cluster.name
+  body_format     = "json"
+  include_headers = false
+}

integration/source.tf

@@ -97,6 +97,13 @@ resource "materialize_source_kafka" "example_source_kafka_format_avro" {
   depends_on = [materialize_sink_kafka.sink_kafka]
 }
 
+resource "materialize_source_webhook" "example_webhook_source" {
+  name            = "example_webhook_source"
+  cluster_name    = materialize_cluster.cluster_source.name
+  body_format     = "json"
+  include_headers = false
+}
+
 resource "materialize_source_grant" "source_grant_select" {
   role_name     = materialize_role.role_1.name
   privilege     = "SELECT"

pkg/materialize/source_webhook.go

@@ -0,0 +1,109 @@
+package materialize
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/jmoiron/sqlx"
+)
+
+type FieldStruct struct {
+	Body    bool
+	Headers bool
+	Secret  IdentifierSchemaStruct
+}
+
+type CheckOptionsStruct struct {
+	Field FieldStruct
+	Alias string
+}
+
+type SourceWebhookBuilder struct {
+	Source
+	clusterName     string
+	size            string
+	bodyFormat      string
+	includeHeaders  bool
+	checkOptions    []CheckOptionsStruct
+	checkExpression string
+}
+
+func NewSourceWebhookBuilder(conn *sqlx.DB, obj ObjectSchemaStruct) *SourceWebhookBuilder {
+	b := Builder{conn, BaseSource}
+	return &SourceWebhookBuilder{
+		Source: Source{b, obj.Name, obj.SchemaName, obj.DatabaseName},
+	}
+}
+
+func (b *SourceWebhookBuilder) ClusterName(c string) *SourceWebhookBuilder {
+	b.clusterName = c
+	return b
+}
+
+func (b *SourceWebhookBuilder) BodyFormat(f string) *SourceWebhookBuilder {
+	b.bodyFormat = f
+	return b
+}
+
+func (b *SourceWebhookBuilder) IncludeHeaders(h bool) *SourceWebhookBuilder {
+	b.includeHeaders = h
+	return b
+}
+
+func (b *SourceWebhookBuilder) CheckOptions(o []CheckOptionsStruct) *SourceWebhookBuilder {
+	b.checkOptions = o
+	return b
+}
+
+func (b *SourceWebhookBuilder) CheckExpression(e string) *SourceWebhookBuilder {
+	b.checkExpression = e
+	return b
+}
+
+func (b *SourceWebhookBuilder) Size(s string) *SourceWebhookBuilder {
+	b.size = s
+	return b
+}
+
+func (b *SourceWebhookBuilder) Create() error {
+	q := strings.Builder{}
+	q.WriteString(fmt.Sprintf(`CREATE SOURCE %s`, b.QualifiedName()))
+	q.WriteString(fmt.Sprintf(` IN CLUSTER %s`, QuoteIdentifier(b.clusterName)))
+	q.WriteString(` FROM WEBHOOK`)
+	q.WriteString(fmt.Sprintf(` BODY FORMAT %s`, b.bodyFormat))
+
+	if b.includeHeaders {
+		q.WriteString(` INCLUDE HEADERS`)
+	}
+
+	if len(b.checkOptions) > 0 || b.checkExpression != "" {
+		q.WriteString(` CHECK (`)
+		if len(b.checkOptions) > 0 {
+			q.WriteString(` WITH (`)
+			var options []string
+			for _, option := range b.checkOptions {
+				if option.Field.Body {
+					options = append(options, "BODY")
+				}
+				if option.Field.Headers {
+					options = append(options, "HEADERS")
+				}
+				if option.Field.Secret.Name != "" {
+					options = append(options, "SECRET "+option.Field.Secret.QualifiedName())
+				}
+				if option.Alias != "" {
+					options[len(options)-1] += " AS " + option.Alias
+				}
+			}
+			q.WriteString(strings.Join(options, ", "))
+			q.WriteString(`) `)
+		}
+		if b.checkExpression != "" {
+			q.WriteString(b.checkExpression)
+		}
+		q.WriteString(`)`)
+	}
+
+	q.WriteString(`;`)
+	return b.ddl.exec(q.String())
+}

pkg/materialize/source_webhook_test.go

@@ -0,0 +1,44 @@
+package materialize
+
+import (
+	"testing"
+
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"github.com/MaterializeInc/terraform-provider-materialize/pkg/testhelpers"
+	"github.com/jmoiron/sqlx"
+)
+
+var sourceWebhook = ObjectSchemaStruct{Name: "webhook_source", SchemaName: "schema", DatabaseName: "database"}
+var checkOptions = []CheckOptionsStruct{
+	{
+		Field: FieldStruct{
+			Body: true,
+		},
+		Alias: "bytes",
+	},
+	{
+		Field: FieldStruct{
+			Headers: true,
+		},
+		Alias: "headers",
+	},
+}
+
+func TestSourceWebhookCreate(t *testing.T) {
+	testhelpers.WithMockDb(t, func(db *sqlx.DB, mock sqlmock.Sqlmock) {
+		mock.ExpectExec(
+			`CREATE SOURCE "database"."schema"."webhook_source" IN CLUSTER "cluster" FROM WEBHOOK BODY FORMAT JSON INCLUDE HEADERS CHECK \( WITH \(BODY AS bytes\, HEADERS AS headers\) check_expression\);`,
+		).WillReturnResult(sqlmock.NewResult(1, 1))
+
+		b := NewSourceWebhookBuilder(db, sourceWebhook)
+		b.ClusterName("cluster")
+		b.BodyFormat("JSON")
+		b.IncludeHeaders(true)
+		b.CheckOptions(checkOptions)
+		b.CheckExpression("check_expression")
+
+		if err := b.Create(); err != nil {
+			t.Fatal(err)
+		}
+	})
+}