uik: Add pending_signals table (#3959)

* Add signal message handling and enforce limits

- Introduced PendingSignal struct and related DB schema
- Defined and imposed new limits for pending signals
- Added new signal message types, enums, and SQL queries
- Implemented action channel verification
- Revised existing code to handle new signal message types and limits
- Compiled and executed conditional rules for actions

* revert version change

* Enhance email and webhook destination configuration

- Added `IsDynamicAction` and `DynamicParams` to email and webhook destinations for flexible parameter handling
- Removed unused `ActionResult` struct from integration key module

* regen

* fix dest creation

* drop funcs on migrate down

* gen

* remove limit rows on migrate down

gadb/dest.go

@@ -57,6 +57,24 @@ type NullDestV1 struct {
 	DestV1 DestV1
 }
 
+func (ns NullDestV1) MarshalJSON() ([]byte, error) {
+	if !ns.Valid {
+		return []byte("null"), nil
+	}
+
+	return json.Marshal(ns.DestV1)
+}
+
+func (ns *NullDestV1) UnmarshalJSON(data []byte) error {
+	if string(data) == "null" {
+		ns.Valid = false
+		return nil
+	}
+
+	ns.Valid = true
+	return json.Unmarshal(data, &ns.DestV1)
+}
+
 // Scan implements the Scanner interface.
 func (ns *NullDestV1) Scan(value interface{}) error {
 	if value == nil {

graphql2/graphqlapp/destinationtypes.go

@@ -312,13 +312,23 @@ func (q *Query) DestinationTypes(ctx context.Context, isDynamicAction *bool) ([]
 			Enabled:               cfg.SMTP.Enable,
 			IsContactMethod:       true,
 			SupportsStatusUpdates: true,
+			IsDynamicAction:       true,
 			RequiredFields: []graphql2.DestinationFieldConfig{{
 				FieldID:            fieldEmailAddress,
 				Label:              "Email Address",
 				PlaceholderText:    "foobar@example.com",
 				InputType:          "email",
 				SupportsValidation: true,
 			}},
+			DynamicParams: []graphql2.DynamicParamConfig{{
+				ParamID: "subject",
+				Label:   "Subject",
+				Hint:    "Subject of the email message.",
+			}, {
+				ParamID: "body",
+				Label:   "Body",
+				Hint:    "Body of the email message.",
+			}},
 		},
 		{
 			Type:                  destWebhook,
@@ -338,6 +348,19 @@ func (q *Query) DestinationTypes(ctx context.Context, isDynamicAction *bool) ([]
 				HintURL:            "/docs#webhooks",
 				SupportsValidation: true,
 			}},
+			IsDynamicAction: true,
+			DynamicParams: []graphql2.DynamicParamConfig{
+				{
+					ParamID: "body",
+					Label:   "Body",
+					Hint:    "The body of the request.",
+				},
+				{
+					ParamID: "content-type",
+					Label:   "Content Type",
+					Hint:    "The content type (usually application/json).",
+				},
+			},
 		},
 		{
 			Type:                  destSlackDM,

integrationkey/config.go

@@ -48,6 +48,8 @@ type Rule struct {
 
 // An Action is a single action to take if a rule matches.
 type Action struct {
+	ChannelID uuid.UUID
+
 	// Type is the type of action to perform, like slack, email, or alert.
 	Type string
 
@@ -167,11 +169,19 @@ func (s *Store) SetConfig(ctx context.Context, db gadb.DBTX, keyID uuid.UUID, cf
 		return gdb.IntKeyDeleteConfig(ctx, keyID)
 	}
 
-	// ensure all rule IDs are set
+	// ensure all rule IDs are set, and all actions have a channel
 	for i := range cfg.Rules {
 		if cfg.Rules[i].ID == uuid.Nil {
 			cfg.Rules[i].ID = uuid.New()
 		}
+		err := setActionChannels(ctx, gdb, cfg.Rules[i].Actions)
+		if err != nil {
+			return err
+		}
+	}
+	err = setActionChannels(ctx, gdb, cfg.DefaultActions)
+	if err != nil {
+		return err
 	}
 
 	data, err := json.Marshal(dbConfig{Version: 1, V1: *cfg})
@@ -189,3 +199,21 @@ func (s *Store) SetConfig(ctx context.Context, db gadb.DBTX, keyID uuid.UUID, cf
 
 	return nil
 }
+
+func setActionChannels(ctx context.Context, gdb *gadb.Queries, actions []Action) error {
+	for j, act := range actions {
+		id, err := gdb.IntKeyEnsureChannel(ctx, gadb.IntKeyEnsureChannelParams{
+			ID: uuid.New(),
+			Dest: gadb.NullDestV1{Valid: true, DestV1: gadb.DestV1{
+				Type: act.Type,
+				Args: act.StaticParams,
+			}},
+		})
+		if err != nil {
+			return err
+		}
+		actions[j].ChannelID = id
+	}
+
+	return nil
+}

integrationkey/queries.sql

@@ -135,3 +135,27 @@ SET
 WHERE
     id = $1;
 
+-- name: IntKeyEnsureChannel :one
+WITH insert_q AS (
+INSERT INTO notification_channels(id, dest, name)
+        VALUES ($1, $2, 'unknown')
+    ON CONFLICT (dest)
+        DO NOTHING
+    RETURNING
+        id)
+    SELECT
+        id
+    FROM
+        insert_q
+    UNION
+    SELECT
+        id
+    FROM
+        notification_channels
+    WHERE
+        dest = $2;
+
+-- name: IntKeyInsertSignalMessage :exec
+INSERT INTO pending_signals(dest_id, service_id, params)
+    VALUES ($1, $2, $3);
+

integrationkey/uik.go

@@ -35,12 +35,6 @@ func newClaims(keyID, tokenID uuid.UUID) jwt.Claims {
 	}
 }
 
-type ActionResult struct {
-	DestType string
-	Values   map[string]string
-	Params   map[string]string
-}
-
 func (s *Store) AuthorizeUIK(ctx context.Context, tokStr string) (context.Context, error) {
 	if !expflag.ContextHas(ctx, expflag.UnivKeys) {
 		return ctx, permission.Unauthorized()

integrationkey/uik/compilerule.go

@@ -0,0 +1,31 @@
+package uik
+
+import (
+	"fmt"
+	"reflect"
+	"strconv"
+	"strings"
+
+	"github.com/expr-lang/expr"
+	"github.com/expr-lang/expr/vm"
+	"github.com/target/goalert/integrationkey"
+)
+
+func CompileRule(cond string, actions []integrationkey.Action) (*vm.Program, error) {
+	var actionExpr []string
+	for _, a := range actions {
+		var params []string
+		for k, v := range a.DynamicParams {
+			params = append(params, fmt.Sprintf(`%s: string(%s)`, strconv.Quote(k), v))
+		}
+
+		actionExpr = append(actionExpr, fmt.Sprintf(`{ %s }`, strings.Join(params, ", ")))
+	}
+
+	src := "[" + strings.Join(actionExpr, ",") + "]"
+	if cond != "" {
+		src = fmt.Sprintf(`string(%s) == "true" ? %s : nil`, cond, src)
+	}
+
+	return expr.Compile(src, expr.AsKind(reflect.Slice), expr.AllowUndefinedVariables(), expr.Optimize(true))
+}