[CAPPL-364] Return an error when secrets are empty (#15635)

core/services/workflows/syncer/handler.go

@@ -456,12 +456,13 @@ func (h *eventHandler) workflowRegisteredEvent(
 	}
 
 	wfID := hex.EncodeToString(payload.WorkflowID[:])
+	owner := hex.EncodeToString(payload.WorkflowOwner)
 	entry := &job.WorkflowSpec{
 		Workflow:      hex.EncodeToString(decodedBinary),
 		Config:        string(config),
 		WorkflowID:    wfID,
 		Status:        status,
-		WorkflowOwner: hex.EncodeToString(payload.WorkflowOwner),
+		WorkflowOwner: owner,
 		WorkflowName:  payload.WorkflowName,
 		SpecType:      job.WASMFile,
 		BinaryURL:     payload.BinaryURL,
@@ -480,7 +481,7 @@ func (h *eventHandler) workflowRegisteredEvent(
 	engine, err := h.engineFactory(
 		ctx,
 		wfID,
-		string(payload.WorkflowOwner),
+		owner,
 		payload.WorkflowName,
 		config,
 		decodedBinary,

core/services/workflows/syncer/orm.go

@@ -161,6 +161,10 @@ func (orm *orm) GetContentsByWorkflowID(ctx context.Context, workflowID string)
 		return "", "", ErrEmptySecrets
 	}
 
+	if jr.Contents.String == "" {
+		return "", "", ErrEmptySecrets
+	}
+
 	return jr.SecretsURLHash.String, jr.Contents.String, nil
 }
 

core/services/workflows/syncer/orm_test.go

@@ -256,3 +256,37 @@ func Test_GetContentsByWorkflowID(t *testing.T) {
 	assert.Equal(t, giveHash, gotHash)
 	assert.Equal(t, giveContent, gotContent)
 }
+
+func Test_GetContentsByWorkflowID_SecretsProvidedButEmpty(t *testing.T) {
+	db := pgtest.NewSqlxDB(t)
+	ctx := testutils.Context(t)
+	lggr := logger.TestLogger(t)
+	orm := &orm{ds: db, lggr: lggr}
+
+	// workflow_id is missing
+	_, _, err := orm.GetContentsByWorkflowID(ctx, "doesnt-exist")
+	require.ErrorContains(t, err, "no rows in result set")
+
+	// secrets_id is nil; should return EmptySecrets
+	workflowID := "aWorkflowID"
+	giveURL := "https://example.com"
+	giveBytes, err := crypto.Keccak256([]byte(giveURL))
+	require.NoError(t, err)
+	giveHash := hex.EncodeToString(giveBytes)
+	giveContent := ""
+	_, err = orm.UpsertWorkflowSpecWithSecrets(ctx, &job.WorkflowSpec{
+		Workflow:      "",
+		Config:        "",
+		WorkflowID:    workflowID,
+		WorkflowOwner: "aWorkflowOwner",
+		WorkflowName:  "aWorkflowName",
+		BinaryURL:     "",
+		ConfigURL:     "",
+		CreatedAt:     time.Now(),
+		SpecType:      job.DefaultSpecType,
+	}, giveURL, giveHash, giveContent)
+	require.NoError(t, err)
+
+	_, _, err = orm.GetContentsByWorkflowID(ctx, workflowID)
+	require.ErrorIs(t, err, ErrEmptySecrets)
+}