feat: call webhooks at least once (#5454)

* feat: call webhooks at least once

* self review

* feat: improve notification observability

* feat: add notification tracing

* test(e2e): test at-least-once webhook delivery

* fix webhook notifications

* dedicated quota notifications handler

* fix linting

* fix e2e test

* wait less in e2e test

* fix: don't ignore failed events in handlers

* fix: don't ignore failed events in handlers

* faster requeues

* question

* fix retries

* fix retries

* retry

* don't instance ids query

* revert handler_projection

* statements can be nil

* cleanup

* make unit tests pass

* add comments

* add comments

* lint

* spool only active instances

* feat(config): handle inactive instances

* customizable HandleInactiveInstances

* call inactive instances quota webhooks

* test: handling with and w/o inactive instances

* omit retrying noop statements

* docs: describe projection options

* enable global handling of inactive instances

* self review

* requeue quota notifications every 5m

* remove caos_errors reference

* fix comment styles

* make handlers package flat

* fix linting

* fix repeating quota notifications

* test with more usage

* debug log channel init failures

internal/notification/channels/channel.go

@@ -1,7 +1,10 @@
 package channels
 
+import "github.com/zitadel/zitadel/internal/eventstore"
+
 type Message interface {
-	GetContent() string
+	GetTriggeringEvent() eventstore.Event
+	GetContent() (string, error)
 }
 
 type NotificationChannel interface {

internal/notification/channels/fs/channel.go

@@ -2,19 +2,16 @@ package fs
 
 import (
 	"fmt"
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"sort"
 	"strings"
 	"time"
 
+	"github.com/k3a/html2text"
 	"github.com/zitadel/logging"
 
-	caos_errors "github.com/zitadel/zitadel/internal/errors"
-
-	"github.com/k3a/html2text"
-
+	"github.com/zitadel/zitadel/internal/errors"
 	"github.com/zitadel/zitadel/internal/notification/channels"
 	"github.com/zitadel/zitadel/internal/notification/messages"
 )
@@ -29,7 +26,10 @@ func InitFSChannel(config Config) (channels.NotificationChannel, error) {
 	return channels.HandleMessageFunc(func(message channels.Message) error {
 
 		fileName := fmt.Sprintf("%d_", time.Now().Unix())
-		content := message.GetContent()
+		content, err := message.GetContent()
+		if err != nil {
+			return err
+		}
 		switch msg := message.(type) {
 		case *messages.Email:
 			recipients := make([]string, len(msg.Recipients))
@@ -41,10 +41,12 @@ func InitFSChannel(config Config) (channels.NotificationChannel, error) {
 			}
 		case *messages.SMS:
 			fileName = fileName + "sms_to_" + msg.RecipientPhoneNumber + ".txt"
+		case *messages.JSON:
+			fileName = "message.json"
 		default:
-			return caos_errors.ThrowUnimplementedf(nil, "NOTIF-6f9a1", "filesystem provider doesn't support message type %T", message)
+			return errors.ThrowUnimplementedf(nil, "NOTIF-6f9a1", "filesystem provider doesn't support message type %T", message)
 		}
 
-		return ioutil.WriteFile(filepath.Join(config.Path, fileName), []byte(content), 0666)
+		return os.WriteFile(filepath.Join(config.Path, fileName), []byte(content), 0666)
 	}), nil
 }

internal/notification/channels/instrumenting/instrument.go

@@ -0,0 +1,26 @@
+package instrumenting
+
+import (
+	"context"
+
+	"github.com/zitadel/zitadel/internal/notification/channels"
+)
+
+func Wrap(
+	ctx context.Context,
+	channel channels.NotificationChannel,
+	traceSpanName,
+	successMetricName,
+	failureMetricName string,
+) channels.NotificationChannel {
+	return traceMessages(
+		ctx,
+		countMessages(
+			ctx,
+			logMessages(ctx, channel),
+			successMetricName,
+			failureMetricName,
+		),
+		traceSpanName,
+	)
+}

internal/notification/channels/instrumenting/logging.go

@@ -0,0 +1,24 @@
+package instrumenting
+
+import (
+	"context"
+
+	"github.com/zitadel/logging"
+
+	"github.com/zitadel/zitadel/internal/api/authz"
+	"github.com/zitadel/zitadel/internal/notification/channels"
+)
+
+func logMessages(ctx context.Context, channel channels.NotificationChannel) channels.NotificationChannel {
+	return channels.HandleMessageFunc(func(message channels.Message) error {
+		logEntry := logging.WithFields(
+			"instance", authz.GetInstance(ctx).InstanceID(),
+			"triggering_event_type", message.GetTriggeringEvent().Type(),
+		)
+		logEntry.Debug("sending notification")
+		err := channel.HandleMessage(message)
+		logEntry.OnError(err).Warn("sending notification failed")
+		logEntry.Debug("notification sent")
+		return err
+	})
+}

internal/notification/channels/instrumenting/metrics.go

@@ -0,0 +1,36 @@
+package instrumenting
+
+import (
+	"context"
+
+	"github.com/zitadel/logging"
+	"go.opentelemetry.io/otel/attribute"
+
+	"github.com/zitadel/zitadel/internal/api/authz"
+	"github.com/zitadel/zitadel/internal/notification/channels"
+	"github.com/zitadel/zitadel/internal/telemetry/metrics"
+)
+
+func countMessages(ctx context.Context, channel channels.NotificationChannel, successMetricName, errorMetricName string) channels.NotificationChannel {
+	return channels.HandleMessageFunc(func(message channels.Message) error {
+		err := channel.HandleMessage(message)
+		metricName := successMetricName
+		if err != nil {
+			metricName = errorMetricName
+		}
+		addCount(ctx, metricName, message, err)
+		return err
+	})
+}
+
+func addCount(ctx context.Context, metricName string, message channels.Message, err error) {
+	labels := map[string]attribute.Value{
+		"triggering_event_typey": attribute.StringValue(string(message.GetTriggeringEvent().Type())),
+		"instance":               attribute.StringValue(authz.GetInstance(ctx).InstanceID()),
+	}
+	if err != nil {
+		labels["error"] = attribute.StringValue(err.Error())
+	}
+	addCountErr := metrics.AddCount(ctx, metricName, 1, labels)
+	logging.WithFields("name", metricName, "labels", labels).OnError(addCountErr).Error("incrementing counter metric failed")
+}

internal/notification/channels/instrumenting/tracing.go

@@ -0,0 +1,16 @@
+package instrumenting
+
+import (
+	"context"
+
+	"github.com/zitadel/zitadel/internal/notification/channels"
+	"github.com/zitadel/zitadel/internal/telemetry/tracing"
+)
+
+func traceMessages(ctx context.Context, channel channels.NotificationChannel, spanName string) channels.NotificationChannel {
+	return channels.HandleMessageFunc(func(message channels.Message) (err error) {
+		_, span := tracing.NewNamedSpan(ctx, spanName)
+		defer func() { span.EndWithError(err) }()
+		return channel.HandleMessage(message)
+	})
+}

internal/notification/channels/log/channel.go

@@ -15,7 +15,10 @@ func InitStdoutChannel(config Config) channels.NotificationChannel {
 
 	return channels.HandleMessageFunc(func(message channels.Message) error {
 
-		content := message.GetContent()
+		content, err := message.GetContent()
+		if err != nil {
+			return err
+		}
 		if config.Compact {
 			content = html2text.HTML2Text(content)
 		}

internal/notification/channels/smtp/channel.go

@@ -22,7 +22,7 @@ type Email struct {
 	senderName    string
 }
 
-func InitSMTPChannel(ctx context.Context, getSMTPConfig func(ctx context.Context) (*Config, error)) (*Email, error) {
+func InitChannel(ctx context.Context, getSMTPConfig func(ctx context.Context) (*Config, error)) (*Email, error) {
 	smtpConfig, err := getSMTPConfig(ctx)
 	if err != nil {
 		return nil, err
@@ -70,7 +70,12 @@ func (email *Email) HandleMessage(message channels.Message) error {
 		return err
 	}
 
-	_, err = w.Write([]byte(emailMsg.GetContent()))
+	content, err := emailMsg.GetContent()
+	if err != nil {
+		return err
+	}
+
+	_, err = w.Write([]byte(content))
 	if err != nil {
 		return err
 	}
@@ -80,7 +85,6 @@ func (email *Email) HandleMessage(message channels.Message) error {
 		return err
 	}
 
-	defer logging.LogWithFields("EMAI-a1c87ec8").Debug("email sent")
 	return email.smtpClient.Quit()
 }
 
@@ -154,6 +158,8 @@ func (smtpConfig SMTP) smtpAuth(client *smtp.Client, host string) error {
 	// Auth
 	auth := smtp.PlainAuth("", smtpConfig.User, smtpConfig.Password, host)
 	err := client.Auth(auth)
-	logging.Log("EMAIL-s9kfs").WithField("smtp user", smtpConfig.User).OnError(err).Debug("could not add smtp auth")
-	return err
+	if err != nil {
+		return caos_errs.ThrowInternalf(err, "EMAIL-s9kfs", "could not add smtp auth for user %s", smtpConfig.User)
+	}
+	return nil
 }

internal/notification/channels/twilio/channel.go

@@ -9,7 +9,7 @@ import (
 	"github.com/zitadel/zitadel/internal/notification/messages"
 )
 
-func InitTwilioChannel(config Config) channels.NotificationChannel {
+func InitChannel(config Config) channels.NotificationChannel {
 	client := twilio.NewClient(config.SID, config.Token, nil)
 
 	logging.Debug("successfully initialized twilio sms channel")
@@ -19,7 +19,11 @@ func InitTwilioChannel(config Config) channels.NotificationChannel {
 		if !ok {
 			return caos_errs.ThrowInternal(nil, "TWILI-s0pLc", "message is not SMS")
 		}
-		m, err := client.Messages.SendMessage(twilioMsg.SenderPhoneNumber, twilioMsg.RecipientPhoneNumber, twilioMsg.GetContent(), nil)
+		content, err := twilioMsg.GetContent()
+		if err != nil {
+			return err
+		}
+		m, err := client.Messages.SendMessage(twilioMsg.SenderPhoneNumber, twilioMsg.RecipientPhoneNumber, content, nil)
 		if err != nil {
 			return caos_errs.ThrowInternal(err, "TWILI-osk3S", "could not send message")
 		}

internal/notification/channels/webhook/channel.go

@@ -0,0 +1,60 @@
+package webhook
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/zitadel/logging"
+
+	"github.com/zitadel/zitadel/internal/errors"
+	"github.com/zitadel/zitadel/internal/notification/channels"
+	"github.com/zitadel/zitadel/internal/notification/messages"
+)
+
+func InitChannel(ctx context.Context, cfg Config) (channels.NotificationChannel, error) {
+	if err := cfg.Validate(); err != nil {
+		return nil, err
+	}
+
+	logging.Debug("successfully initialized webhook json channel")
+	return channels.HandleMessageFunc(func(message channels.Message) error {
+
+		requestCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+		defer cancel()
+
+		msg, ok := message.(*messages.JSON)
+		if !ok {
+			return errors.ThrowInternal(nil, "WEBH-K686U", "message is not JSON")
+		}
+		payload, err := msg.GetContent()
+		if err != nil {
+			return err
+		}
+
+		req, err := http.NewRequestWithContext(requestCtx, cfg.Method, cfg.CallURL, strings.NewReader(payload))
+		if err != nil {
+			return err
+		}
+
+		req.Header.Set("Content-Type", "application/json")
+
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return err
+		}
+
+		if err = resp.Body.Close(); err != nil {
+			return err
+		}
+
+		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+			return errors.ThrowUnknown(fmt.Errorf("calling url %s returned %s", cfg.CallURL, resp.Status), "WEBH-LBxU0", "webhook didn't return a success status")
+		}
+
+		logging.WithFields("calling_url", cfg.CallURL, "method", cfg.Method).Debug("webhook called")
+		return nil
+	}), nil
+}

internal/notification/channels/webhook/config.go

@@ -0,0 +1,15 @@
+package webhook
+
+import (
+	"net/url"
+)
+
+type Config struct {
+	CallURL string
+	Method  string
+}
+
+func (w *Config) Validate() error {
+	_, err := url.Parse(w.CallURL)
+	return err
+}