Feat: dynamic crons fixes (#1073)

hatchet-dev · Nov 26, 2024 · 2320166 · 2320166
1 parent 8da776f
commit 2320166
Show file tree

Hide file tree

Showing 12 changed files with 565 additions and 512 deletions.
diff --git a/api-contracts/openapi/paths/workflow/workflow.yaml b/api-contracts/openapi/paths/workflow/workflow.yaml
@@ -843,24 +843,12 @@ scheduledCreate:
           format: uuid
           minLength: 36
           maxLength: 36
-      - description: The workflow id
+      - description: The workflow name
         in: path
         name: workflow
         required: true
         schema:
           type: string
-          format: uuid
-          minLength: 36
-          maxLength: 36
-      - description: The workflow version. If not supplied, the latest version is fetched.
-        in: query
-        name: version
-        required: false
-        schema:
-          type: string
-          format: uuid
-          minLength: 36
-          maxLength: 36
     requestBody:
       content:
         application/json:
@@ -1201,7 +1189,7 @@ crons:
 
 cronsCreate:
   post:
-    x-resources: ["tenant", "workflow"]
+    x-resources: ["tenant"]
     description: Create a new cron job workflow trigger for a tenant
     operationId: cron-workflow-trigger:create
     parameters:
@@ -1214,24 +1202,12 @@ cronsCreate:
           format: uuid
           minLength: 36
           maxLength: 36
-      - description: The workflow id
+      - description: The workflow name
         in: path
         name: workflow
         required: true
         schema:
           type: string
-          format: uuid
-          minLength: 36
-          maxLength: 36
-      - description: The workflow version. If not supplied, the latest version is fetched.
-        in: query
-        name: version
-        required: false
-        schema:
-          type: string
-          format: uuid
-          minLength: 36
-          maxLength: 36
     requestBody:
       content:
         application/json:

diff --git a/api/v1/server/handlers/workflows/create_cron.go b/api/v1/server/handlers/workflows/create_cron.go
@@ -1,13 +1,16 @@
 package workflows
 
 import (
+	"strings"
+
 	"github.com/labstack/echo/v4"
 
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/apierrors"
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/gen"
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/transformers"
 	"github.com/hatchet-dev/hatchet/pkg/repository"
 	"github.com/hatchet-dev/hatchet/pkg/repository/prisma/db"
+	"github.com/hatchet-dev/hatchet/pkg/repository/prisma/sqlchelpers"
 )
 
 func (t *WorkflowService) CronWorkflowTriggerCreate(ctx echo.Context, request gen.CronWorkflowTriggerCreateRequestObject) (gen.CronWorkflowTriggerCreateResponseObject, error) {
@@ -17,18 +20,28 @@ func (t *WorkflowService) CronWorkflowTriggerCreate(ctx echo.Context, request ge
 		return gen.CronWorkflowTriggerCreate400JSONResponse(apierrors.NewAPIErrors("cron name is required")), nil
 	}
 
+	workflow, err := t.config.EngineRepository.Workflow().GetWorkflowByName(ctx.Request().Context(), tenant.ID, request.Workflow)
+
+	if err != nil {
+		return gen.CronWorkflowTriggerCreate400JSONResponse(apierrors.NewAPIErrors("workflow not found")), nil
+	}
+
 	cronTrigger, err := t.config.APIRepository.Workflow().CreateCronWorkflow(
 		ctx.Request().Context(), tenant.ID, &repository.CreateCronWorkflowTriggerOpts{
 			Name:               request.Body.CronName,
 			Cron:               request.Body.CronExpression,
 			Input:              request.Body.Input,
 			AdditionalMetadata: request.Body.AdditionalMetadata,
-			WorkflowId:         request.Workflow.String(),
+			WorkflowId:         sqlchelpers.UUIDToStr(workflow.ID),
 		},
 	)
 
 	if err != nil {
-		return gen.CronWorkflowTriggerCreate400JSONResponse(apierrors.NewAPIErrors(err.Error())), nil
+		if strings.Contains(err.Error(), "unique constraint") {
+			return gen.CronWorkflowTriggerCreate400JSONResponse(apierrors.NewAPIErrors("cron trigger with that name-expression pair already exists")), nil
+		}
+
+		return gen.CronWorkflowTriggerCreate400JSONResponse(apierrors.NewAPIErrors("error creating cron trigger")), nil
 	}
 
 	return gen.CronWorkflowTriggerCreate200JSONResponse(

diff --git a/api/v1/server/handlers/workflows/create_scheduled.go b/api/v1/server/handlers/workflows/create_scheduled.go
@@ -8,16 +8,23 @@ import (
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/transformers"
 	"github.com/hatchet-dev/hatchet/pkg/repository"
 	"github.com/hatchet-dev/hatchet/pkg/repository/prisma/db"
+	"github.com/hatchet-dev/hatchet/pkg/repository/prisma/sqlchelpers"
 )
 
 func (t *WorkflowService) ScheduledWorkflowRunCreate(ctx echo.Context, request gen.ScheduledWorkflowRunCreateRequestObject) (gen.ScheduledWorkflowRunCreateResponseObject, error) {
 	tenant := ctx.Get("tenant").(*db.TenantModel)
 
+	workflow, err := t.config.EngineRepository.Workflow().GetWorkflowByName(ctx.Request().Context(), tenant.ID, request.Workflow)
+
+	if err != nil {
+		return gen.ScheduledWorkflowRunCreate400JSONResponse(apierrors.NewAPIErrors("workflow not found")), nil
+	}
+
 	scheduled, err := t.config.APIRepository.Workflow().CreateScheduledWorkflow(ctx.Request().Context(), tenant.ID, &repository.CreateScheduledWorkflowRunForWorkflowOpts{
 		ScheduledTrigger:   request.Body.TriggerAt,
 		Input:              request.Body.Input,
 		AdditionalMetadata: request.Body.AdditionalMetadata,
-		WorkflowId:         request.Workflow.String(),
+		WorkflowId:         sqlchelpers.UUIDToStr(workflow.ID),
 	})
 
 	if err != nil {