Merge pull request #507 from serverlessworkflow/fix-function-call

Fixed the `FunctionCallExecutor` to support custom functions that do not specify an @ catalog, falling back to the default Serverless Workflow catalog

Author: cdavernas

src/runner/Synapse.Runner/Program.cs

@@ -11,6 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+using Json.Schema;
 using Moq;
 using Neuroglia.AsyncApi;
 using Neuroglia.AsyncApi.Client;
@@ -129,4 +130,17 @@
 
 using var app = builder.Build();
 
+SchemaRegistry.Global.Fetch = uri =>
+{
+    using var scope = app.Services.CreateScope();
+    using var client = scope.ServiceProvider.GetRequiredService<IHttpClientFactory>().CreateClient();
+    using var request = new HttpRequestMessage(HttpMethod.Get, uri);
+    using var response = client.Send(request);
+    response.EnsureSuccessStatusCode();
+    using var stream = response.Content.ReadAsStream();
+    var contentType = response.Content.Headers.ContentType?.MediaType!;
+    var serializer = scope.ServiceProvider.GetRequiredService<ISerializerProvider>().GetSerializersFor(contentType).FirstOrDefault() ?? throw new NullReferenceException($"Failed to find a serializer for the specified content type '{contentType}'");
+    return serializer.Deserialize<JsonSchema>(stream);
+};
+
 await app.RunAsync();

src/runner/Synapse.Runner/Services/Executors/FunctionCallExecutor.cs

@@ -35,8 +35,8 @@ public class FunctionCallExecutor(IServiceProvider serviceProvider, ILogger<Func
 {
 
     const string CustomFunctionDefinitionFile = "function.yaml";
-    const string GithubHost = "github.com";
-    const string GitlabHost = "gitlab";
+    const string GitHubHost = "github.com";
+    const string GitLabHost = "gitlab";
 
     /// <summary>
     /// Gets the service used to serialize/deserialize objects to/from YAML
@@ -65,6 +65,15 @@ public override async Task InitializeAsync(CancellationToken cancellationToken =
             if (components.Length != 2) throw new NotSupportedException($"Unknown/unsupported function '{this.Task.Definition.Call}'");
             this.Function = await this.GetCustomFunctionAsync(components[0], components[1], cancellationToken).ConfigureAwait(false);
         }
+        else if (this.Task.Definition.Call.Contains(':'))
+        {
+            var components = this.Task.Definition.Call.Split(':', StringSplitOptions.RemoveEmptyEntries);
+            if (components.Length != 2) throw new Exception($"The specified value '{this.Task.Definition.Call}' is not a valid custom function qualified name ({{name}}:{{version}})");
+            var functionName = components[0];
+            var functionVersion = components[1];
+            uri = new Uri($"https://github.com/serverlessworkflow/catalog/tree/main/functions/{functionName}/{functionVersion}/{CustomFunctionDefinitionFile}");
+            this.Function = await this.GetCustomFunctionAsync(new() { Uri = uri }, cancellationToken).ConfigureAwait(false);
+        }
         else throw new NotSupportedException($"Unknown/unsupported function '{this.Task.Definition.Call}'");
     }
 
@@ -79,8 +88,8 @@ protected virtual async Task<TaskDefinition> GetCustomFunctionAsync(EndpointDefi
         ArgumentNullException.ThrowIfNull(endpoint);
         var uri = endpoint.Uri;
         if (!uri.OriginalString.EndsWith(CustomFunctionDefinitionFile)) uri = new Uri(uri, CustomFunctionDefinitionFile);
-        if (uri.Host.Equals(GithubHost, StringComparison.OrdinalIgnoreCase)) uri = this.TransformGithubUriToRawUri(uri);
-        else if (uri.Host.Contains(GitlabHost)) uri = this.TransformGitlabUriToRawUri(uri);
+        if (uri.Host.Equals(GitHubHost, StringComparison.OrdinalIgnoreCase)) uri = this.TransformGithubUriToRawUri(uri);
+        else if (uri.Host.Contains(GitLabHost)) uri = this.TransformGitlabUriToRawUri(uri);
         var authentication = endpoint.Authentication == null ? null : await this.Task.Workflow.Expressions.EvaluateAsync<AuthenticationPolicyDefinition>(endpoint.Authentication, this.Task.Input, this.Task.Arguments, cancellationToken).ConfigureAwait(false);
         using var httpClient = this.ServiceProvider.GetRequiredService<IHttpClientFactory>().CreateClient();
         await httpClient.ConfigureAuthenticationAsync(authentication, this.ServiceProvider, this.Task.Workflow.Definition, cancellationToken).ConfigureAwait(false);
@@ -138,8 +147,8 @@ protected virtual async Task<TaskDefinition> GetCustomFunctionAsync(string funct
     protected virtual Uri TransformGithubUriToRawUri(Uri uri)
     {
         ArgumentNullException.ThrowIfNull(uri);
-        if (uri.Host.Equals(GithubHost, StringComparison.OrdinalIgnoreCase)) return uri;
-        var rawUri = uri.AbsoluteUri.Replace(GithubHost, "raw.githubusercontent.com", StringComparison.OrdinalIgnoreCase);
+        if (!uri.Host.Equals(GitHubHost, StringComparison.OrdinalIgnoreCase)) return uri;
+        var rawUri = uri.AbsoluteUri.Replace(GitHubHost, "raw.githubusercontent.com", StringComparison.OrdinalIgnoreCase);
         rawUri = rawUri.Replace("/tree/", "/refs/heads/", StringComparison.OrdinalIgnoreCase);
         return new(rawUri, UriKind.Absolute);
     }
@@ -152,7 +161,7 @@ protected virtual Uri TransformGithubUriToRawUri(Uri uri)
     protected virtual Uri TransformGitlabUriToRawUri(Uri uri)
     {
         ArgumentNullException.ThrowIfNull(uri);
-        if (!uri.AbsoluteUri.Contains(GitlabHost, StringComparison.OrdinalIgnoreCase)) return uri;
+        if (!uri.Host.Equals(GitLabHost, StringComparison.OrdinalIgnoreCase)) return uri;
         var rawUri = uri.AbsoluteUri.Replace("/-/blob/", "/-/raw/", StringComparison.OrdinalIgnoreCase);
         return new(rawUri, UriKind.Absolute);
     }

src/runner/Synapse.Runner/Services/Executors/HttpCallExecutor.cs

@@ -104,7 +104,7 @@ protected override async Task DoExecuteAsync(CancellationToken cancellationToken
         {
             requestContent = this.Http.Body switch
             {
-                string stringContent => new StringContent(stringContent, Encoding.UTF8, mediaType),
+                string stringContent => stringContent.IsRuntimeExpression() ? null : new StringContent(stringContent, Encoding.UTF8, mediaType),
                 byte[] byteArrayContent => new StreamContent(new MemoryStream(byteArrayContent)),
                 _ => null
             };