From 600126a913163a17a9a3a1759a7587c98e8d3a48 Mon Sep 17 00:00:00 2001 From: mrhid6 Date: Mon, 20 Jul 2026 11:33:33 +0100 Subject: [PATCH] feat(server): step library and workflow CRUD services --- server/cmd/main.go | 4 + server/internal/services/workflows.go | 170 ++++++++++++++++++++++++++ 2 files changed, 174 insertions(+) create mode 100644 server/internal/services/workflows.go diff --git a/server/cmd/main.go b/server/cmd/main.go index 41ac110..6f71e21 100644 --- a/server/cmd/main.go +++ b/server/cmd/main.go @@ -27,6 +27,10 @@ func main() { log.Printf("warning: failed to ensure secret indexes: %v", err) } + if err := services.EnsureWorkflowIndexes(); err != nil { + log.Printf("warning: failed to ensure workflow indexes: %v", err) + } + redisAddr := getEnv("REDIS_ADDR", "localhost:6379") if err := auth.InitRedis(redisAddr); err != nil { log.Fatalf("failed to connect to Redis: %v", err) diff --git a/server/internal/services/workflows.go b/server/internal/services/workflows.go new file mode 100644 index 0000000..b7c2b7b --- /dev/null +++ b/server/internal/services/workflows.go @@ -0,0 +1,170 @@ +package services + +import ( + "context" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/mrhid6/vantage/server/internal/db" + "github.com/mrhid6/vantage/server/internal/models" + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" +) + +func wfCtx() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), 10*time.Second) +} + +func EnsureWorkflowIndexes() error { + ctx, cancel := wfCtx() + defer cancel() + if _, err := db.Col("workflow_steps").Indexes().CreateOne(ctx, mongo.IndexModel{ + Keys: bson.D{{Key: "step_id", Value: 1}}, Options: options.Index().SetUnique(true), + }); err != nil { + return err + } + if _, err := db.Col("workflows").Indexes().CreateOne(ctx, mongo.IndexModel{ + Keys: bson.D{{Key: "workflow_id", Value: 1}}, Options: options.Index().SetUnique(true), + }); err != nil { + return err + } + _, err := db.Col("workflow_runs").Indexes().CreateOne(ctx, mongo.IndexModel{ + Keys: bson.D{{Key: "run_id", Value: 1}}, Options: options.Index().SetUnique(true), + }) + return err +} + +// ---- Steps ---- + +func ListSteps() ([]models.WorkflowStep, error) { + ctx, cancel := wfCtx() + defer cancel() + cur, err := db.Col("workflow_steps").Find(ctx, bson.M{}, + options.Find().SetSort(bson.D{{Key: "name", Value: 1}})) + if err != nil { + return nil, err + } + defer cur.Close(ctx) + steps := []models.WorkflowStep{} + if err := cur.All(ctx, &steps); err != nil { + return nil, err + } + return steps, nil +} + +func CreateStep(s models.WorkflowStep) (*models.WorkflowStep, error) { + ctx, cancel := wfCtx() + defer cancel() + s.StepID = uuid.New().String() + s.CreatedAt = time.Now() + s.UpdatedAt = s.CreatedAt + if s.DeclaredOutputs == nil { + s.DeclaredOutputs = []string{} + } + if s.SecretRefs == nil { + s.SecretRefs = []string{} + } + if _, err := db.Col("workflow_steps").InsertOne(ctx, s); err != nil { + return nil, err + } + return &s, nil +} + +func UpdateStep(stepID string, s models.WorkflowStep) error { + ctx, cancel := wfCtx() + defer cancel() + _, err := db.Col("workflow_steps").UpdateOne(ctx, bson.M{"step_id": stepID}, bson.M{"$set": bson.M{ + "name": s.Name, + "description": s.Description, + "interpreter": s.Interpreter, + "script": s.Script, + "declared_outputs": s.DeclaredOutputs, + "secret_refs": s.SecretRefs, + "updated_at": time.Now(), + }}) + return err +} + +func DeleteStep(stepID string) error { + ctx, cancel := wfCtx() + defer cancel() + _, err := db.Col("workflow_steps").DeleteOne(ctx, bson.M{"step_id": stepID}) + return err +} + +func getStep(ctx context.Context, stepID string) (*models.WorkflowStep, error) { + var s models.WorkflowStep + err := db.Col("workflow_steps").FindOne(ctx, bson.M{"step_id": stepID}).Decode(&s) + if err == mongo.ErrNoDocuments { + return nil, fmt.Errorf("step %s not found", stepID) + } + return &s, err +} + +// ---- Workflows ---- + +func ListWorkflows() ([]models.Workflow, error) { + ctx, cancel := wfCtx() + defer cancel() + cur, err := db.Col("workflows").Find(ctx, bson.M{}, + options.Find().SetSort(bson.D{{Key: "name", Value: 1}})) + if err != nil { + return nil, err + } + defer cur.Close(ctx) + wfs := []models.Workflow{} + if err := cur.All(ctx, &wfs); err != nil { + return nil, err + } + return wfs, nil +} + +func GetWorkflow(id string) (*models.Workflow, error) { + ctx, cancel := wfCtx() + defer cancel() + var w models.Workflow + err := db.Col("workflows").FindOne(ctx, bson.M{"workflow_id": id}).Decode(&w) + if err == mongo.ErrNoDocuments { + return nil, fmt.Errorf("workflow not found") + } + return &w, err +} + +func CreateWorkflow(w models.Workflow) (*models.Workflow, error) { + ctx, cancel := wfCtx() + defer cancel() + w.WorkflowID = uuid.New().String() + w.CreatedAt = time.Now() + w.UpdatedAt = w.CreatedAt + if w.TargetServerIDs == nil { + w.TargetServerIDs = []string{} + } + if w.Steps == nil { + w.Steps = []models.WorkflowStepRef{} + } + if _, err := db.Col("workflows").InsertOne(ctx, w); err != nil { + return nil, err + } + return &w, nil +} + +func UpdateWorkflow(id string, w models.Workflow) error { + ctx, cancel := wfCtx() + defer cancel() + _, err := db.Col("workflows").UpdateOne(ctx, bson.M{"workflow_id": id}, bson.M{"$set": bson.M{ + "name": w.Name, + "target_server_ids": w.TargetServerIDs, + "steps": w.Steps, + "updated_at": time.Now(), + }}) + return err +} + +func DeleteWorkflow(id string) error { + ctx, cancel := wfCtx() + defer cancel() + _, err := db.Col("workflows").DeleteOne(ctx, bson.M{"workflow_id": id}) + return err +}