Added sub-workflows to notifications

Author: hughneale

internal/common/client_test.go

@@ -0,0 +1,35 @@
+package common
+
+import (
+	"crypto/sha256"
+	"testing"
+
+	"github.com/google/uuid"
+)
+
+func TestGetClientIdentifier_UsesEnvUUID(t *testing.T) {
+	expected := uuid.New()
+	t.Setenv(clientIdentifierEnvVar, expected.String())
+
+	got := GetClientIdentifier()
+	if got != expected {
+		t.Fatalf("GetClientIdentifier() = %s, expected %s", got.String(), expected.String())
+	}
+}
+
+func TestGetClientIdentifier_UsesEnvStringHash(t *testing.T) {
+	value := "not-a-uuid-but-still-unique"
+	t.Setenv(clientIdentifierEnvVar, value)
+
+	got1 := GetClientIdentifier()
+	got2 := GetClientIdentifier()
+	if got1 != got2 {
+		t.Fatalf("expected deterministic UUID for env override; got %s then %s", got1.String(), got2.String())
+	}
+
+	hash := sha256.Sum256([]byte(value))
+	expected := uuid.UUID(hash[:16])
+	if got1 != expected {
+		t.Fatalf("GetClientIdentifier() = %s, expected %s", got1.String(), expected.String())
+	}
+}

internal/config/providers.go

@@ -10,6 +10,7 @@ import (
 	"github.com/thand-io/agent/internal/config/environment"
 	"github.com/thand-io/agent/internal/models"
 	"github.com/thand-io/agent/internal/providers"
+	sdkConstants "github.com/thand-io/agent/sdk/constants"
 	providerSdk "github.com/thand-io/agent/sdk/providers"
 	"go.temporal.io/sdk/workflow"
 
@@ -190,7 +191,7 @@ func (c *Config) InitializeProviders() error {
 	if c.IsServer() {
 		_ = c.GetServices()
 		if err := c.SetupTemporal(); err != nil {
-			return fmt.Errorf("setting up temporal services: %w", err)
+			return fmt.Errorf("failed to set up temporal services: %w", err)
 		}
 	}
 
@@ -239,25 +240,29 @@ func (c *Config) InitializeProviders() error {
 			models.ProviderCapabilityRoles,
 			models.ProviderCapabilityPermissions,
 			models.ProviderCapabilityTenants,
+			models.ProviderCapabilityNotifier,
+			models.ProviderCapabilityProvisioning,
 		) {
 
 			logrus.Infoln("Provider", result.key, "supports RBAC/Identities capabilities")
 
-			// Register provider workflows and activities with Temporal if available
-			if c.IsServer() {
+			// If a server or agent register all capabilities depending on their runtime
 
-				if c.GetServices() != nil && c.GetServices().HasTemporal() {
+			if c.GetServices() != nil && c.GetServices().HasTemporal() {
 
-					logrus.Infoln("Registering Temporal workflows/activities for provider", result.key)
+				logrus.Infoln("Registering Temporal workflows/activities for provider", result.key)
 
-					temporalService := c.GetServices().GetTemporal()
+				temporalService := c.GetServices().GetTemporal()
 
-					worker := temporalService.GetWorker()
+				worker := temporalService.GetWorker()
 
-					if worker == nil {
-						logrus.Errorln("Temporal client is configured but worker is nil, cannot register workflows/activities for provider", result.key)
-						continue
-					}
+				if worker == nil {
+					logrus.Errorln("Temporal client is configured but worker is nil, cannot register workflows/activities for provider", result.key)
+					continue
+				}
+
+				// Only sync jobs can happen on the server
+				if c.IsServer() {
 
 					syncWorkflowName := models.CreateTemporalProviderWorkflowName(
 						providerResult.GetIdentifier(),
@@ -277,75 +282,146 @@ func (c *Config) InitializeProviders() error {
 							VersioningBehavior: workflow.VersioningBehaviorPinned,
 						},
 					)
+				}
+
+				logrus.Info("Registering default activities for provider", providerResult.GetName())
+
+				// Check the provider capability is either agent or server
+				providerCapabilities := providerResult.GetCapabilities()
 
-					if providerResult.HasCapability(models.ProviderCapabilityProvisioning) {
-
-						authWorkflowName := models.CreateTemporalProviderWorkflowName(
-							providerResult.GetIdentifier(),
-							models.TemporalAuthorizeRoleWorkflowName)
-
-						logrus.WithFields(logrus.Fields{
-							"workflow": authWorkflowName,
-							"provider": providerResult.GetIdentifier(),
-						}).Infoln("Registering provider authorize role workflow with name", authWorkflowName)
-
-						// Register the provider-specific authorize and revoke role workflows.
-						// These are closure-based: they capture the live provider instance so the
-						// child workflow can call provider.AuthorizeRole / RevokeRole with a
-						// full workflow.Context, allowing providers to dispatch activities,
-						// use workflow.Go, etc.
-						worker.RegisterWorkflowWithOptions(
-							models.CreateProviderAuthorizeRoleWorkflow(providerResult),
-							workflow.RegisterOptions{
-								Name:               authWorkflowName,
-								VersioningBehavior: workflow.VersioningBehaviorPinned,
-							},
-						)
-
-						revokeWorkflowName := models.CreateTemporalProviderWorkflowName(
-							providerResult.GetIdentifier(),
-							models.TemporalRevokeRoleWorkflowName)
-
-						logrus.WithFields(logrus.Fields{
-							"workflow": revokeWorkflowName,
-							"provider": providerResult.GetIdentifier(),
-						}).Infoln("Registering provider revoke role workflow with name", revokeWorkflowName)
-
-						worker.RegisterWorkflowWithOptions(
-							models.CreateProviderRevokeRoleWorkflow(providerResult),
-							workflow.RegisterOptions{
-								Name:               revokeWorkflowName,
-								VersioningBehavior: workflow.VersioningBehaviorPinned,
-							},
-						)
+				// We always register the provider on the server
+				// so that it can re-set the task queue if needed.
+				// only register on agent/client if its needed.
+
+				// Register the provisioning capability
+				if providerCapabilities != nil &&
+					providerCapabilities.Provisioning.Enabled &&
+					(providerCapabilities.Provisioning.Runtime == c.GetMode() || (providerCapabilities.Provisioning.Runtime == sdkConstants.ModeAgent && c.IsClient()) || c.IsServer()) {
+
+					authWorkflowName := models.CreateTemporalProviderWorkflowName(
+						providerResult.GetIdentifier(),
+						models.TemporalAuthorizeRoleWorkflowName)
+
+					logrus.WithFields(logrus.Fields{
+						"workflow": authWorkflowName,
+						"provider": providerResult.GetIdentifier(),
+					}).Infoln("Registering provider authorize role workflow with name", authWorkflowName)
+
+					// Register the provider-specific authorize and revoke role workflows.
+					// These are closure-based: they capture the live provider instance so the
+					// child workflow can call provider.AuthorizeRole / RevokeRole with a
+					// full workflow.Context, allowing providers to dispatch activities,
+					// use workflow.Go, etc.
+					worker.RegisterWorkflowWithOptions(
+						models.CreateProviderAuthorizeRoleWorkflow(providerResult),
+						workflow.RegisterOptions{
+							Name:               authWorkflowName,
+							VersioningBehavior: workflow.VersioningBehaviorPinned,
+						},
+					)
+
+					revokeWorkflowName := models.CreateTemporalProviderWorkflowName(
+						providerResult.GetIdentifier(),
+						models.TemporalRevokeRoleWorkflowName)
+
+					logrus.WithFields(logrus.Fields{
+						"workflow": revokeWorkflowName,
+						"provider": providerResult.GetIdentifier(),
+					}).Infoln("Registering provider revoke role workflow with name", revokeWorkflowName)
+
+					worker.RegisterWorkflowWithOptions(
+						models.CreateProviderRevokeRoleWorkflow(providerResult),
+						workflow.RegisterOptions{
+							Name:               revokeWorkflowName,
+							VersioningBehavior: workflow.VersioningBehaviorPinned,
+						},
+					)
+				} else {
+					provisioningEnabled := false
+					provisioningRuntime := sdkConstants.Mode("")
+					if providerCapabilities != nil && providerCapabilities.Provisioning != nil {
+						provisioningEnabled = providerCapabilities.Provisioning.Enabled
+						provisioningRuntime = providerCapabilities.Provisioning.Runtime
 					}
-					// Register all custom provider workflows
-					workflowsRegistry := providerResult.RegisterWorkflows()
-					if workflowsRegistry != nil {
-						logrus.Infoln("Registering Temporal workflows for provider", result.key)
-						worker.RegisterWorkflow(workflowsRegistry)
+					logrus.WithFields(logrus.Fields{
+						"provider":             result.key,
+						"mode":                 c.GetMode(),
+						"provisioning.enabled": provisioningEnabled,
+						"provisioning.runtime": provisioningRuntime,
+					}).Infoln("Skipping provisioning workflow registration: provider does not support provisioning in this runtime")
+				}
+
+				// Register the notification capability
+				if providerCapabilities != nil &&
+					providerCapabilities.Notifier.Enabled &&
+					((providerCapabilities.Notifier.Runtime == c.GetMode()) || (providerCapabilities.Notifier.Runtime == sdkConstants.ModeAgent && c.IsClient()) || c.IsServer()) {
+
+					notifierWorkflowName := models.CreateTemporalProviderWorkflowName(
+						providerResult.GetIdentifier(),
+						models.TemporalNotifyWorkflowName)
+
+					logrus.WithFields(logrus.Fields{
+						"workflow": notifierWorkflowName,
+						"provider": providerResult.GetIdentifier(),
+					}).Infoln("Registering provider notify workflow with name", notifierWorkflowName)
+
+					worker.RegisterWorkflowWithOptions(
+						models.CreateProviderNotifyWorkflow(providerResult),
+						workflow.RegisterOptions{
+							Name:               notifierWorkflowName,
+							VersioningBehavior: workflow.VersioningBehaviorPinned,
+						},
+					)
+				} else {
+					notifierEnabled := false
+					notifierRuntime := sdkConstants.Mode("")
+					if providerCapabilities != nil && providerCapabilities.Notifier != nil {
+						notifierEnabled = providerCapabilities.Notifier.Enabled
+						notifierRuntime = providerCapabilities.Notifier.Runtime
 					}
+					logrus.WithFields(logrus.Fields{
+						"provider":         result.key,
+						"mode":             c.GetMode(),
+						"notifier.enabled": notifierEnabled,
+						"notifier.runtime": notifierRuntime,
+					}).Infoln("Skipping notify workflow registration: provider does not support notifier in this runtime")
+				}
+
+				// Register all custom provider workflows
+				workflowsRegistry := providerResult.RegisterWorkflows(c.GetMode())
+				if workflowsRegistry != nil {
+					logrus.Infoln("Registering Temporal workflows for provider", result.key)
+					worker.RegisterWorkflow(workflowsRegistry)
+				}
+
+				logrus.Infoln("Finished registering Temporal workflows/activities for provider", result.key)
+
+				// Register default provider activities
+				err := models.RegisterProviderActivities(temporalService, providerResult, c)
+				if err != nil {
+					logrus.WithError(err).Errorln("Failed to register default activities for provider:", result.key)
+					continue
+				}
+
+				logrus.Infoln("Registered default activities for provider", providerResult.GetName())
 
-					// Register default provider activities
-					err := models.RegisterProviderActivities(temporalService, providerResult, c)
+				customActivities := providerResult.RegisterActivities(c.GetMode())
+				if customActivities != nil {
+
+					logrus.Infoln("Registering custom Temporal activities for provider", result.key)
+
+					// Now register any custom activities defined by the provider
+					err = models.RegisterActivities(
+						temporalService,
+						providerResult.GetIdentifier(),
+						customActivities,
+					)
 					if err != nil {
-						logrus.WithError(err).Errorln("Failed to register default activities for provider:", result.key)
+						logrus.WithError(err).Errorln("Failed to register custom activities for provider:", result.key)
 						continue
 					}
-
-					customActivities := providerResult.RegisterActivities()
-					if customActivities != nil {
-						// Now register any custom activities defined by the provider
-						err = models.RegisterActivities(
-							temporalService,
-							providerResult.GetIdentifier(),
-							customActivities,
-						)
-						if err != nil {
-							logrus.WithError(err).Errorln("Failed to register custom activities for provider:", result.key)
-							continue
-						}
-					}
+				} else {
+					logrus.Infoln("No custom activities to register for provider", result.key)
 				}
 
 				logrus.Infoln("Synchronizing provider", result.key)
@@ -358,7 +434,7 @@ func (c *Config) InitializeProviders() error {
 			}
 		} else {
 			// Provider doesn't have RBAC/Identity capabilities, no sync needed
-			result.provider.SetReady()
+			providerResult.SetReady()
 		}
 
 		// The provider returned from the goroutine already has the client set
@@ -425,7 +501,8 @@ func (c *Config) initializeSingleProvider(providerKey string, p *models.Provider
 // getProviderImplementation returns the appropriate provider implementation based on config mode
 func (c *Config) getProviderImplementation(providerKey string, providerName string) (models.Provider, error) {
 
-	if c.IsServer() || c.IsAgent() {
+	// TODO
+	if c.IsServer() || c.IsAgent() || c.IsClient() {
 		return providers.CreateInstance(strings.ToLower(providerName))
 	}
 

internal/daemon/model.go

@@ -9,6 +9,7 @@ type SimpleServices struct {
 
 type SimpleConfig struct {
 	ApiBasePath string
+	Mode        string
 	Server      SimpleServer
 	Services    SimpleServices
 }

internal/daemon/server.go

@@ -156,6 +156,7 @@ func (s *Server) GetTemplateData(c *gin.Context) TemplateData {
 	return TemplateData{
 		Config: SimpleConfig{
 			ApiBasePath: s.Config.GetApiBasePath(),
+			Mode:        string(s.Config.GetMode()),
 			Server: SimpleServer{
 				Host: s.Config.Server.Host,
 				Port: s.Config.Server.Port,

internal/models/provider.go

@@ -10,6 +10,7 @@ import (
 	"github.com/hashicorp/go-version"
 	"github.com/thand-io/agent/internal/common"
 	"github.com/thand-io/agent/internal/interpolate"
+	sdkConstants "github.com/thand-io/agent/sdk/constants"
 )
 
 var ErrNotImplemented = errors.New("not implemented")
@@ -111,8 +112,8 @@ type Provider interface {
 	Synchronize(ctx context.Context, temporalClient TemporalImpl, req *SynchronizeRequest) error
 
 	// Temporal
-	RegisterWorkflows() any
-	RegisterActivities() any
+	RegisterWorkflows(runtime sdkConstants.Mode) any
+	RegisterActivities(runtime sdkConstants.Mode) any
 
 	GetCapabilities() *ProviderCapabilities
 	HasCapability(capability ProviderCapability) bool

internal/models/provider_activities.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 
 	"github.com/sirupsen/logrus"
+	sdkConstants "github.com/thand-io/agent/sdk/constants"
 	"go.temporal.io/sdk/activity"
 	"go.temporal.io/sdk/temporal"
 )
@@ -16,7 +17,7 @@ import (
 // To expose additional, provider-specific activities, override RegisterActivities
 // on your provider struct to return a populated activities struct (or nil to skip):
 //
-//	func (p *myProvider) RegisterActivities() any {
+//	func (p *myProvider) RegisterActivities(runtime sdkConstants.Mode) any {
 //	    return &myProviderActivities{provider: p}
 //	}
 //
@@ -68,7 +69,7 @@ func RegisterProviderActivities(temporalClient TemporalImpl, provider Provider,
 }
 
 // RegisterActivities — BaseProvider default; returns ErrNotImplemented.
-func (b *BaseProvider) RegisterActivities() any {
+func (b *BaseProvider) RegisterActivities(runtime sdkConstants.Mode) any {
 	return nil
 }