Flexible schema support with default fallback in Email Executor

backend/internal/flow/common/constants.go

@@ -193,6 +193,8 @@ const (
 	RuntimeKeyUserAttributesCacheTTLSeconds = "user_attributes_cache_ttl_seconds"
 	// RuntimeKeyInviteLink holds the generated invite link for downstream executors (e.g., EmailExecutor).
 	RuntimeKeyInviteLink = "inviteLink"
+	// RuntimeKeySkipDelivery indicates that delivery should be skipped for the current flow.
+	RuntimeKeySkipDelivery = "skipDelivery"
 	// RuntimeKeyCandidateUsers holds serialized candidate users during disambiguation in resolve mode.
 	RuntimeKeyCandidateUsers = "candidateUsers"
 )
@@ -203,6 +205,8 @@ const (
 const (
 	// InputTypeText represents a text input type.
 	InputTypeText = "TEXT_INPUT"
+	// InputTypeEmail represents an email input type.
+	InputTypeEmail = "EMAIL_INPUT"
 	// InputTypePassword represents a password credential input type.
 	InputTypePassword = "PASSWORD_INPUT"
 	// InputTypeOTP represents a one-time password input type.

backend/internal/flow/executor/email_executor.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/asgardeo/thunder/internal/entityprovider"
 	"github.com/asgardeo/thunder/internal/flow/common"
 	"github.com/asgardeo/thunder/internal/flow/core"
 	"github.com/asgardeo/thunder/internal/system/email"
@@ -36,25 +37,36 @@ type emailExecutor struct {
 	logger          *log.Logger
 	emailClient     email.EmailClientInterface
 	templateService template.TemplateServiceInterface
+	entityProvider  entityprovider.EntityProviderInterface
+}
+
+// defaultEmailInput is the default input definition for email collection.
+var defaultEmailInput = common.Input{
+	Ref:        "email_input",
+	Identifier: userAttributeEmail,
+	Type:       common.InputTypeEmail,
+	Required:   true,
 }
 
 // newEmailExecutor creates a new instance of the email executor.
 func newEmailExecutor(flowFactory core.FlowFactoryInterface, emailClient email.EmailClientInterface,
-	templateService template.TemplateServiceInterface) *emailExecutor {
+	templateService template.TemplateServiceInterface,
+	entityProvider entityprovider.EntityProviderInterface) *emailExecutor {
 	logger := log.GetLogger().With(log.String(log.LoggerKeyComponentName, "EmailExecutor"))
 	base := flowFactory.CreateExecutor(
 		ExecutorNameEmailExecutor,
 		common.ExecutorTypeUtility,
 		[]common.Input{},
 		[]common.Input{
-			{Identifier: userAttributeEmail, Type: common.InputTypeText, Required: true},
+			defaultEmailInput,
 		},
 	)
 	return &emailExecutor{
 		ExecutorInterface: base,
 		logger:            logger,
 		emailClient:       emailClient,
 		templateService:   templateService,
+		entityProvider:    entityProvider,
 	}
 }
 
@@ -79,6 +91,12 @@ func (e *emailExecutor) executeSend(ctx *core.NodeContext) (*common.ExecutorResp
 		RuntimeData:    make(map[string]string),
 	}
 
+	if ctx.RuntimeData[common.RuntimeKeySkipDelivery] == dataValueTrue {
+		logger.Debug("Delivery marked as skipped, completing without sending email")
+		execResp.Status = common.ExecComplete
+		return execResp, nil
+	}
+
 	if e.emailClient == nil {
 		execResp.AdditionalData[common.DataEmailSent] = dataValueFalse
 		execResp.Status = common.ExecFailure
@@ -92,7 +110,10 @@ func (e *emailExecutor) executeSend(ctx *core.NodeContext) (*common.ExecutorResp
 	}
 
 	// Resolve recipient email from user inputs or runtime data.
-	recipient := resolveRecipientEmail(ctx)
+	recipient, err := e.resolveRecipientEmail(ctx, logger)
+	if err != nil {
+		return nil, err
+	}
 	if recipient == "" {
 		logger.Debug("Email recipient not found in user inputs or runtime data")
 		execResp.Status = common.ExecFailure
@@ -125,7 +146,6 @@ func (e *emailExecutor) executeSend(ctx *core.NodeContext) (*common.ExecutorResp
 		"inviteLink": inviteLink,
 		"appName":    ctx.Application.Name,
 	}
-
 	rendered, svcErr := e.templateService.Render(ctx.Context, scenario, template.TemplateTypeEmail, templateData)
 	if svcErr != nil {
 		return nil, fmt.Errorf("failed to render email template: %s", svcErr.Code)
@@ -156,15 +176,41 @@ func (e *emailExecutor) executeSend(ctx *core.NodeContext) (*common.ExecutorResp
 	return execResp, nil
 }
 
-// resolveRecipientEmail retrieves the recipient email from user inputs or runtime data.
-func resolveRecipientEmail(ctx *core.NodeContext) string {
-	if recipientEmail, ok := ctx.UserInputs[userAttributeEmail]; ok && recipientEmail != "" {
-		return recipientEmail
+// resolveRecipientEmail retrieves the recipient email from user inputs, runtime data, or forwarded data.
+func (e *emailExecutor) resolveRecipientEmail(ctx *core.NodeContext, logger *log.Logger) (string, error) {
+	emailAttr := e.resolveEmailInput(ctx).Identifier
+
+	if recipientEmail, ok := ctx.ForwardedData[emailAttr]; ok {
+		if emailStr, isString := recipientEmail.(string); isString && emailStr != "" {
+			return emailStr, nil
+		}
 	}
-	if recipientEmail, ok := ctx.RuntimeData[userAttributeEmail]; ok && recipientEmail != "" {
-		return recipientEmail
+	if recipientEmail, ok := ctx.RuntimeData[emailAttr]; ok && recipientEmail != "" {
+		return recipientEmail, nil
 	}
-	return ""
+	if recipientEmail, ok := ctx.UserInputs[emailAttr]; ok && recipientEmail != "" {
+		return recipientEmail, nil
+	}
+
+	if userID, ok := ctx.RuntimeData[userAttributeUserID]; ok && userID != "" {
+		if e.entityProvider == nil {
+			return "", errors.New("entity provider is not configured for email resolution")
+		}
+		user, providerErr := e.entityProvider.GetEntity(userID)
+		if providerErr != nil {
+			if providerErr.Code == entityprovider.ErrorCodeEntityNotFound {
+				return "", nil
+			}
+			return "", fmt.Errorf("failed to fetch user from entity provider: %w", providerErr)
+		}
+
+		if recipientEmail, err := GetUserAttribute(user, emailAttr); err == nil {
+			return recipientEmail, nil
+		}
+		logger.Debug("Email attribute not found in user entity", log.String("attribute", emailAttr))
+	}
+
+	return "", nil
 }
 
 // isEmailError returns true if the error originated from the email subsystem,
@@ -180,3 +226,14 @@ func isEmailError(err error) bool {
 		errors.Is(err, email.ErrorSMTPAuth) ||
 		errors.Is(err, email.ErrorEmailSendFailed)
 }
+
+// resolveEmailInput returns the EMAIL_INPUT definition from the node context inputs,
+// falling back to the default if none is found.
+func (e *emailExecutor) resolveEmailInput(ctx *core.NodeContext) common.Input {
+	for _, input := range ctx.NodeInputs {
+		if input.Type == common.InputTypeEmail {
+			return input
+		}
+	}
+	return defaultEmailInput
+}