chore: workflows engine working e2e

apps/workspace-engine/main.go

@@ -41,7 +41,7 @@ func main() {
 
 	allServices := []svc.Service{
 		pprof.New(pprof.DefaultAddr(config.Global.PprofPort)),
-		httpsvc.New(config.Global),
+		httpsvc.New(config.Global, db.GetPool(ctx)),
 		claimcleanup.New(db.GetPool(ctx), 30*time.Second),
 
 		deploymentplan.New(WorkerID, db.GetPool(ctx)),

apps/workspace-engine/oapi/openapi.json

@@ -2933,9 +2933,6 @@
                   "description": "Configuration for the job agent",
                   "type": "object"
                },
-               "id": {
-                  "type": "string"
-               },
                "name": {
                   "type": "string"
                },
@@ -2949,7 +2946,6 @@
                }
             },
             "required": [
-               "id",
                "name",
                "ref",
                "config",

apps/workspace-engine/oapi/spec/schemas/workflows.jsonnet

@@ -3,10 +3,9 @@ local openapi = import '../lib/openapi.libsonnet';
 {
   WorkflowJobAgent: {
     type: 'object',
-    required: ['id', 'name', 'ref', 'config', 'selector'],
+    required: ['name', 'ref', 'config', 'selector'],
     properties: {
       name: { type: 'string' },
-      id: { type: 'string' },
       ref: { type: 'string', description: 'Reference to the job agent' },
       config: { type: 'object', additionalProperties: true, description: 'Configuration for the job agent' },
       selector: { type: 'string', description: 'CEL expression to determine if the job agent should dispatch a job' },

apps/workspace-engine/pkg/oapi/persistence.go

@@ -78,7 +78,7 @@ func (wt *Workflow) CompactionKey() (string, string) {
 }
 
 func (wtt *WorkflowJobAgent) CompactionKey() (string, string) {
-	return "workflow_job_agent", wtt.Id
+	return "workflow_job_agent", wtt.Ref
 }
 
 func (w *WorkflowRun) CompactionKey() (string, string) {

apps/workspace-engine/svc/controllers/jobdispatch/setters_postgres.go

@@ -99,10 +99,12 @@ func (s *PostgresSetter) UpdateJob(
 	}
 
 	releaseID := existingJob.ReleaseId
-	if releaseID != "" {
-		if err := dispatchProgressionTargets(ctx, s.Queue, jobIDUUID); err != nil {
-			return fmt.Errorf("dispatch progression targets: %w", err)
-		}
+	if releaseID == "" || releaseID == "00000000-0000-0000-0000-000000000000" {
+		return nil
+	}
+
+	if err := dispatchProgressionTargets(ctx, s.Queue, jobIDUUID); err != nil {
+		return fmt.Errorf("dispatch progression targets: %w", err)
 	}
 
 	return nil

apps/workspace-engine/svc/http/http.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 
 	"github.com/charmbracelet/log"
+	"github.com/jackc/pgx/v5/pgxpool"
 	"workspace-engine/pkg/config"
 	"workspace-engine/svc"
 	"workspace-engine/svc/http/server"
@@ -16,17 +17,18 @@ var _ svc.Service = (*Service)(nil)
 // Service wraps the workspace-engine HTTP server as a service.Service.
 type Service struct {
 	cfg        config.Config
+	pool       *pgxpool.Pool
 	httpServer *http.Server
 }
 
-func New(cfg config.Config) *Service {
-	return &Service{cfg: cfg}
+func New(cfg config.Config, pool *pgxpool.Pool) *Service {
+	return &Service{cfg: cfg, pool: pool}
 }
 
 func (s *Service) Name() string { return "http" }
 
 func (s *Service) Start(_ context.Context) error {
-	srv := server.New()
+	srv := server.New(s.pool)
 	router := srv.SetupRouter()
 
 	addr := fmt.Sprintf("%s:%d", s.cfg.Host, s.cfg.Port)

apps/workspace-engine/svc/http/server/openapi/server.go

@@ -1,6 +1,7 @@
 package openapi
 
 import (
+	"github.com/jackc/pgx/v5/pgxpool"
 	"workspace-engine/pkg/oapi"
 	release_targets "workspace-engine/svc/http/server/openapi/release_targets"
 	"workspace-engine/svc/http/server/openapi/resources"
@@ -9,9 +10,9 @@ import (
 	"workspace-engine/svc/http/server/openapi/workflows"
 )
 
-func New() *Server {
+func New(pool *pgxpool.Pool) *Server {
 	return &Server{
-		Workflows:      workflows.NewWorkflows(),
+		Workflows:      workflows.NewWorkflows(pool),
 		ReleaseTargets: release_targets.New(),
 		Verifications:  verifications.New(),
 	}

apps/workspace-engine/svc/http/server/openapi/workflows/setters.go

@@ -103,7 +103,7 @@ func (s *PostgresSetter) dispatchJobForAgent(
 	dispatchContext *oapi.DispatchContext,
 	workspaceID string,
 ) error {
-	jobAgentIDUUID, err := uuid.Parse(jobAgent.Id)
+	jobAgentIDUUID, err := uuid.Parse(jobAgent.Ref)
 	if err != nil {
 		return fmt.Errorf("parse job agent id: %w", err)
 	}

apps/workspace-engine/svc/http/server/openapi/workflows/workflows.go

@@ -6,16 +6,22 @@ import (
 	"net/http"
 
 	"github.com/gin-gonic/gin"
+	"github.com/jackc/pgx/v5/pgxpool"
 	"workspace-engine/pkg/oapi"
+	"workspace-engine/pkg/reconcile/postgres"
 )
 
 type Workflows struct {
 	getter Getter
 	setter Setter
 }
 
-func NewWorkflows() Workflows {
-	return Workflows{getter: &PostgresGetter{}}
+func NewWorkflows(pool *pgxpool.Pool) Workflows {
+	queue := postgres.New(pool)
+	return Workflows{
+		getter: &PostgresGetter{},
+		setter: NewPostgresSetter(queue),
+	}
 }
 
 func getInputs(c *gin.Context) (map[string]any, error) {

apps/workspace-engine/svc/http/server/openapi/workflows/workflows_test.go

@@ -0,0 +1,176 @@
+package workflows
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"workspace-engine/pkg/oapi"
+)
+
+func stringInput(key string, def *string) oapi.WorkflowInput {
+	var input oapi.WorkflowInput
+	_ = input.FromWorkflowStringInput(oapi.WorkflowStringInput{
+		Key:     key,
+		Type:    "string",
+		Default: def,
+	})
+	return input
+}
+
+func numberInput(key string, def *float32) oapi.WorkflowInput {
+	var input oapi.WorkflowInput
+	_ = input.FromWorkflowNumberInput(oapi.WorkflowNumberInput{
+		Key:     key,
+		Type:    "number",
+		Default: def,
+	})
+	return input
+}
+
+func booleanInput(key string, def *bool) oapi.WorkflowInput {
+	var input oapi.WorkflowInput
+	_ = input.FromWorkflowBooleanInput(oapi.WorkflowBooleanInput{
+		Key:     key,
+		Type:    "boolean",
+		Default: def,
+	})
+	return input
+}
+
+func ptr[T any](v T) *T { return &v }
+
+func TestResolveInputs_ProvidedInputsPassThrough(t *testing.T) {
+	workflow := &oapi.Workflow{
+		Inputs: []oapi.WorkflowInput{
+			stringInput("env", ptr("staging")),
+		},
+	}
+	provided := map[string]any{"env": "production"}
+
+	resolved, err := resolveInputs(workflow, provided)
+
+	require.NoError(t, err)
+	assert.Equal(t, "production", resolved["env"])
+}
+
+func TestResolveInputs_MissingInputsGetDefaults(t *testing.T) {
+	workflow := &oapi.Workflow{
+		Inputs: []oapi.WorkflowInput{
+			stringInput("env", ptr("staging")),
+			numberInput("retries", ptr(float32(3))),
+			booleanInput("dryRun", ptr(true)),
+		},
+	}
+	provided := map[string]any{}
+
+	resolved, err := resolveInputs(workflow, provided)
+
+	require.NoError(t, err)
+	assert.Equal(t, "staging", resolved["env"])
+	assert.Equal(t, float32(3), resolved["retries"])
+	assert.Equal(t, true, resolved["dryRun"])
+}
+
+func TestResolveInputs_InputsWithoutDefaultsStayAbsent(t *testing.T) {
+	workflow := &oapi.Workflow{
+		Inputs: []oapi.WorkflowInput{
+			stringInput("env", nil),
+			numberInput("retries", nil),
+			booleanInput("dryRun", nil),
+		},
+	}
+	provided := map[string]any{}
+
+	resolved, err := resolveInputs(workflow, provided)
+
+	require.NoError(t, err)
+	assert.NotContains(t, resolved, "env")
+	assert.NotContains(t, resolved, "retries")
+	assert.NotContains(t, resolved, "dryRun")
+}
+
+func TestResolveInputs_ProvidedOverridesDefault(t *testing.T) {
+	workflow := &oapi.Workflow{
+		Inputs: []oapi.WorkflowInput{
+			stringInput("env", ptr("staging")),
+			numberInput("retries", ptr(float32(3))),
+			booleanInput("dryRun", ptr(true)),
+		},
+	}
+	provided := map[string]any{
+		"env":     "production",
+		"retries": 10,
+		"dryRun":  false,
+	}
+
+	resolved, err := resolveInputs(workflow, provided)
+
+	require.NoError(t, err)
+	assert.Equal(t, "production", resolved["env"])
+	assert.Equal(t, 10, resolved["retries"])
+	assert.Equal(t, false, resolved["dryRun"])
+}
+
+func TestResolveInputs_MixedProvidedAndDefaults(t *testing.T) {
+	workflow := &oapi.Workflow{
+		Inputs: []oapi.WorkflowInput{
+			stringInput("env", ptr("staging")),
+			numberInput("retries", ptr(float32(3))),
+			booleanInput("verbose", nil),
+		},
+	}
+	provided := map[string]any{"env": "production"}
+
+	resolved, err := resolveInputs(workflow, provided)
+
+	require.NoError(t, err)
+	assert.Equal(t, "production", resolved["env"])
+	assert.Equal(t, float32(3), resolved["retries"])
+	assert.NotContains(t, resolved, "verbose")
+}
+
+func TestResolveInputs_DoesNotMutateProvidedMap(t *testing.T) {
+	workflow := &oapi.Workflow{
+		Inputs: []oapi.WorkflowInput{
+			stringInput("env", ptr("staging")),
+		},
+	}
+	provided := map[string]any{"existing": "value"}
+
+	_, err := resolveInputs(workflow, provided)
+
+	require.NoError(t, err)
+	assert.Len(t, provided, 1)
+	assert.Equal(t, "value", provided["existing"])
+}
+
+func TestResolveInputs_EmptyWorkflowInputs(t *testing.T) {
+	workflow := &oapi.Workflow{
+		Inputs: []oapi.WorkflowInput{},
+	}
+	provided := map[string]any{"extra": "value"}
+
+	resolved, err := resolveInputs(workflow, provided)
+
+	require.NoError(t, err)
+	assert.Equal(t, "value", resolved["extra"])
+}
+
+func TestResolveInputs_ExtraProvidedInputsPassThrough(t *testing.T) {
+	workflow := &oapi.Workflow{
+		Inputs: []oapi.WorkflowInput{
+			stringInput("env", ptr("staging")),
+		},
+	}
+	provided := map[string]any{
+		"env":   "production",
+		"extra": "unexpected",
+	}
+
+	resolved, err := resolveInputs(workflow, provided)
+
+	require.NoError(t, err)
+	assert.Equal(t, "production", resolved["env"])
+	assert.Equal(t, "unexpected", resolved["extra"])
+}

apps/workspace-engine/svc/http/server/server.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/charmbracelet/log"
 	"github.com/gin-gonic/gin"
+	"github.com/jackc/pgx/v5/pgxpool"
 	swaggerfiles "github.com/swaggo/files"
 	ginswagger "github.com/swaggo/gin-swagger"
 	"go.opentelemetry.io/otel"
@@ -20,11 +21,13 @@ import (
 var tracer = otel.Tracer("server")
 
 // Server implements the OpenAPI ServerInterface for the workspace engine.
-type Server struct{}
+type Server struct {
+	pool *pgxpool.Pool
+}
 
 // New creates a new Server instance.
-func New() *Server {
-	return &Server{}
+func New(pool *pgxpool.Pool) *Server {
+	return &Server{pool: pool}
 }
 
 // SetupRouter configures and returns a Gin router with all routes and middleware.
@@ -52,7 +55,7 @@ func (s *Server) SetupRouter() *gin.Engine {
 	)
 
 	// Register OpenAPI handlers
-	oapi.RegisterHandlers(router, openapi.New())
+	oapi.RegisterHandlers(router, openapi.New(s.pool))
 
 	return router
 }