Dynamic schema support and DB fallback to Email Executor

Signed-off-by: RandithaK <me@randitha.net>

Author: RandithaK

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.
@@ -224,6 +228,10 @@ const (
 	// AttributeMobileNumber is the default attribute name for a user's mobile phone number.
 	AttributeMobileNumber = "mobileNumber"
 )
+const (
+	// AttributeEmail is the default attribute name for a user's email.
+	AttributeEmail = "email"
+)
 
 // sensitiveInputTypes contains the list of input types that are considered sensitive.
 var sensitiveInputTypes = []string{

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,11 +37,21 @@ 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: common.AttributeEmail,
+	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,
@@ -55,6 +66,7 @@ func newEmailExecutor(flowFactory core.FlowFactoryInterface, emailClient email.E
 		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,7 @@ func (e *emailExecutor) executeSend(ctx *core.NodeContext) (*common.ExecutorResp
 	}
 
 	// Resolve recipient email from user inputs or runtime data.
-	recipient := resolveRecipientEmail(ctx)
+	recipient := e.resolveRecipientEmail(ctx)
 	if recipient == "" {
 		logger.Debug("Email recipient not found in user inputs or runtime data")
 		execResp.Status = common.ExecFailure
@@ -125,7 +143,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,14 +173,36 @@ 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 != "" {
+// resolveRecipientEmail retrieves the recipient email from user inputs, runtime data, or forwarded data.
+func (e *emailExecutor) resolveRecipientEmail(ctx *core.NodeContext) string {
+	emailAttr := e.resolveEmailInput(ctx).Identifier
+
+	if recipientEmail, ok := ctx.UserInputs[emailAttr]; ok && recipientEmail != "" {
 		return recipientEmail
 	}
-	if recipientEmail, ok := ctx.RuntimeData[userAttributeEmail]; ok && recipientEmail != "" {
+	if recipientEmail, ok := ctx.RuntimeData[emailAttr]; ok && recipientEmail != "" {
 		return recipientEmail
 	}
+	if recipientEmail, ok := ctx.ForwardedData[emailAttr]; ok {
+		if emailStr, isString := recipientEmail.(string); isString && emailStr != "" {
+			return emailStr
+		}
+	}
+
+	if userID, ok := ctx.RuntimeData[userAttributeUserID]; ok && userID != "" && e.entityProvider != nil {
+		user, providerErr := e.entityProvider.GetEntity(userID)
+		if providerErr != nil {
+			e.logger.Error("Failed to fetch user from entity provider for email resolution", log.Error(providerErr))
+		} else if user != nil {
+			if recipientEmail, err := GetUserAttribute(user, emailAttr); err == nil {
+				return recipientEmail
+			} else {
+				e.logger.Debug("Email attribute not found in user entity", log.String("attribute", emailAttr),
+					log.Error(err))
+			}
+		}
+	}
+
 	return ""
 }
 
@@ -180,3 +219,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
+}