feat: add metadata to pubsub notifications

docs/notifications.md

@@ -32,3 +32,13 @@ necessary permissions to publish messages to the configured topic.
 The payload of the messages is the same as that [documented for the `generic`
 destination
 type](https://developer.hashicorp.com/terraform/cloud-docs/api-docs/notification-configurations#run-notification-payload) (using the JSON format).
+
+Additionally, attributes are added to each message:
+
+|key|value|
+|-|-|
+|`otf.ninja/v1/workspace.name`|`<workspace_name>`|
+|`otf.ninja/v1/workspace.id`|`<workspace_id>`|
+|`otf.ninja/v1/tags/<tag_name>`|`true`|
+
+Attributes permit you to [filter messages from a subscription](https://cloud.google.com/pubsub/docs/subscription-message-filter#filtering_syntax) in GCP.

internal/integration/notification_gcppubsub_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/leg100/otf/internal"
 	"github.com/leg100/otf/internal/notifications"
 	"github.com/leg100/otf/internal/testutils"
+	"github.com/leg100/otf/internal/workspace"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -44,6 +45,12 @@ func TestIntegration_NotificationGCPPubSub(t *testing.T) {
 	daemon, _, ctx := setup(t, nil)
 
 	ws := daemon.createWorkspace(t, ctx, nil)
+
+	// add some tags to the workspace so we can check below that they are added
+	// to the pubsub message.
+	err = daemon.AddTags(ctx, ws.ID, []workspace.TagSpec{{Name: "foo"}, {Name: "bar"}})
+	require.NoError(t, err)
+
 	_, err = daemon.CreateNotificationConfiguration(ctx, ws.ID, notifications.CreateConfigOptions{
 		DestinationType: notifications.DestinationGCPPubSub,
 		Enabled:         internal.Bool(true),
@@ -61,8 +68,8 @@ func TestIntegration_NotificationGCPPubSub(t *testing.T) {
 	run := daemon.createRun(t, ctx, ws, cv)
 
 	// gcp-pubsub messages are not necessarily received in the same order as
-	// they are sent, so wait til all three expected messages are received and
-	// then check them.
+	// they are sent, so wait til all expected messages are received and then
+	// check them.
 	var got []*pubsub.Message
 	got = append(got, <-received)
 	got = append(got, <-received)
@@ -84,6 +91,14 @@ func TestIntegration_NotificationGCPPubSub(t *testing.T) {
 			assert.Equal(t, run.ID, payload.RunID)
 			matches[status] = true
 		}
+		// check attributes include workspace metadata
+		want := map[string]string{
+			"otf.ninja/v1/workspace.name": ws.Name,
+			"otf.ninja/v1/workspace.id":   ws.ID,
+			"otf.ninja/v1/tags/foo":       "true",
+			"otf.ninja/v1/tags/bar":       "true",
+		}
+		assert.Equal(t, want, g.Attributes)
 	}
 	// check everything matched
 	for _, want := range matches {

internal/notifications/client_gcppubsub.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/url"
 	"regexp"
 
@@ -70,6 +71,7 @@ func newPubSubClient(cfg *Config) (*pubsubClient, error) {
 	}, nil
 }
 
+// Publish a notification to a gcp pub/sub topic.
 func (c *pubsubClient) Publish(ctx context.Context, n *notification) error {
 	payload, err := n.genericPayload()
 	if err != nil {
@@ -79,8 +81,22 @@ func (c *pubsubClient) Publish(ctx context.Context, n *notification) error {
 	if err != nil {
 		return err
 	}
+
+	// add workspace metadata to allow subscribers to filter messages:
+	//
+	// https://cloud.google.com/pubsub/docs/subscription-message-filter#filtering_syntax
+	attrs := map[string]string{
+		"otf.ninja/v1/workspace.name": n.workspace.Name,
+		"otf.ninja/v1/workspace.id":   n.workspace.ID,
+	}
+	for _, tag := range n.workspace.Tags {
+		key := fmt.Sprintf("otf.ninja/v1/tags/%s", tag)
+		attrs[key] = "true"
+	}
+
 	c.topic.Publish(ctx, &pubsub.Message{
-		Data: data,
+		Attributes: attrs,
+		Data:       data,
 	})
 	return nil
 }