feat(server): step library and workflow CRUD services

2026-07-20 11:33:33 +01:00
parent 4872a26786
commit 600126a913
2 changed files with 174 additions and 0 deletions
@@ -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
+}