108 lines
2.5 KiB
Go
108 lines
2.5 KiB
Go
|
package composition
|
||
|
|
|
||
|
|
import (
|
||
|
|
"fmt"
|
||
|
|
"sync"
|
||
|
|
)
|
||
|
|
|
||
|
|
// ChildOrchestrator represents a child orchestrator workflow
|
||
|
|
type ChildOrchestrator struct {
|
||
|
|
ID string
|
||
|
|
ParentTask string
|
||
|
|
Config map[string]interface{}
|
||
|
|
Status string
|
||
|
|
Results map[string]interface{}
|
||
|
|
CreatedAt int64
|
||
|
|
}
|
||
|
|
|
||
|
|
// WorkflowComposer manages nested orchestrator workflows
|
||
|
|
type WorkflowComposer struct {
|
||
|
|
mu sync.RWMutex
|
||
|
|
children map[string]*ChildOrchestrator
|
||
|
|
results map[string]map[string]interface{}
|
||
|
|
}
|
||
|
|
|
||
|
|
// NewWorkflowComposer creates a new workflow composer
|
||
|
|
func NewWorkflowComposer() *WorkflowComposer {
|
||
|
|
return &WorkflowComposer{
|
||
|
|
children: make(map[string]*ChildOrchestrator),
|
||
|
|
results: make(map[string]map[string]interface{}),
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// CreateChild creates a child orchestrator
|
||
|
|
func (wc *WorkflowComposer) CreateChild(parentTask string, config map[string]interface{}) (*ChildOrchestrator, error) {
|
||
|
|
if parentTask == "" {
|
||
|
|
return nil, fmt.Errorf("parent task required")
|
||
|
|
}
|
||
|
|
|
||
|
|
wc.mu.Lock()
|
||
|
|
defer wc.mu.Unlock()
|
||
|
|
|
||
|
|
child := &ChildOrchestrator{
|
||
|
|
ID: fmt.Sprintf("child-%s-%d", parentTask, len(wc.children)),
|
||
|
|
ParentTask: parentTask,
|
||
|
|
Config: config,
|
||
|
|
Status: "pending",
|
||
|
|
Results: make(map[string]interface{}),
|
||
|
|
}
|
||
|
|
|
||
|
|
wc.children[child.ID] = child
|
||
|
|
return child, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
// GetChild retrieves a child orchestrator
|
||
|
|
func (wc *WorkflowComposer) GetChild(id string) (*ChildOrchestrator, bool) {
|
||
|
|
wc.mu.RLock()
|
||
|
|
defer wc.mu.RUnlock()
|
||
|
|
|
||
|
|
child, exists := wc.children[id]
|
||
|
|
return child, exists
|
||
|
|
}
|
||
|
|
|
||
|
|
// ListChildren lists all children
|
||
|
|
func (wc *WorkflowComposer) ListChildren() map[string]*ChildOrchestrator {
|
||
|
|
wc.mu.RLock()
|
||
|
|
defer wc.mu.RUnlock()
|
||
|
|
|
||
|
|
result := make(map[string]*ChildOrchestrator)
|
||
|
|
for id, child := range wc.children {
|
||
|
|
result[id] = child
|
||
|
|
}
|
||
|
|
|
||
|
|
return result
|
||
|
|
}
|
||
|
|
|
||
|
|
// SetChildStatus updates child status
|
||
|
|
func (wc *WorkflowComposer) SetChildStatus(id string, status string) error {
|
||
|
|
wc.mu.Lock()
|
||
|
|
defer wc.mu.Unlock()
|
||
|
|
|
||
|
|
child, exists := wc.children[id]
|
||
|
|
if !exists {
|
||
|
|
return fmt.Errorf("child not found: %s", id)
|
||
|
|
}
|
||
|
|
|
||
|
|
child.Status = status
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
// GetHierarchy returns the workflow hierarchy
|
||
|
|
func (wc *WorkflowComposer) GetHierarchy() map[string]interface{} {
|
||
|
|
wc.mu.RLock()
|
||
|
|
defer wc.mu.RUnlock()
|
||
|
|
|
||
|
|
children := make([]map[string]interface{}, 0)
|
||
|
|
for _, child := range wc.children {
|
||
|
|
children = append(children, map[string]interface{}{
|
||
|
|
"id": child.ID,
|
||
|
|
"parent_task": child.ParentTask,
|
||
|
|
"status": child.Status,
|
||
|
|
})
|
||
|
|
}
|
||
|
|
|
||
|
|
return map[string]interface{}{
|
||
|
|
"children": children,
|
||
|
|
}
|
||
|
|
}
|