w4c-workflows-api/Services/Nodes/Executors/HttpRequestNodeExecutor.cs

using System.Text.Json.Nodes;
using w4c_workflows.Models.Nodes;
using w4c_workflows.Services.Credentials;
using w4c_workflows.Services.Nodes.Binary;
using w4c_workflows.Services.Security;

namespace w4c_workflows.Services.Nodes.Executors;

/// <summary>
/// HTTP Request node. Calls an endpoint per input item and maps the response to
/// items. Non-2xx responses fail the node (routable to the error port) unless
/// <c>neverError</c> is set; JSON arrays become one item per element.
///
/// Every outbound target — the initial URL and each redirect hop — is vetted by
/// <see cref="EgressGuard"/> before a request is sent, so a workflow cannot reach
/// private/loopback/link-local addresses. Automatic redirects are disabled at the
/// transport and followed here instead, which is what makes per-hop vetting
/// possible; credentials are dropped when a redirect crosses to another host.
///
/// <see cref="NodeRequestBudget"/> additionally bounds the run: each hop consumes
/// one request from the per-run budget, and the response body is streamed under
/// a size cap so an oversized payload fails the node instead of filling memory.
///
/// When the node declares a <c>options.pagination</c> plan the same request is
/// issued repeatedly, driven by <see cref="HttpPaginationPlan"/>: the next URL or
/// cursor is read from a JSON path in the previous response, and the accumulated
/// pages are emitted as one item array. Every page is vetted by the egress guard
/// and consumes quota exactly like a first request, so pagination cannot be used
/// to reach a blocked target or escape the request budget.
///
/// <c>options.responseFormat: file</c> streams the response into the
/// <see cref="IBinaryStore"/> and emits an item whose <c>binary.data</c> reference
/// carries the file name, MIME type and size. Request bodies may be
/// <c>multipart</c> (declared fields plus one input binary attachment) or
/// <c>binary</c> (the input attachment as the raw body), so a download can be
/// fed straight back into an upload without inlining bytes in item JSON.
///
/// The class is split across partial files by responsibility:
/// <c>HttpRequestNodeExecutor.Sending.cs</c> (one request + redirects),
/// <c>HttpRequestNodeExecutor.Body.cs</c> (request body/headers),
/// <c>HttpRequestNodeExecutor.Pagination.cs</c> (page stepping) and
/// <c>HttpRequestNodeExecutor.Response.cs</c> (response → items).
/// </summary>
public sealed partial class HttpRequestNodeExecutor : INodeExecutor
{
    /// <summary>Credential alias declared on the blueprint.</summary>
    public const string CredentialAlias = "httpAuth";

    /// <summary>Blueprint type and default HTTP client name.</summary>
    public const string TypeName = "core.httpRequest";

    /// <summary>Named client used when a node opts out of TLS validation.</summary>
    public const string InsecureClientName = TypeName + ":insecure";

    /// <summary>Default input/output binary property name.</summary>
    public const string DefaultBinaryProperty = FileSystemBinaryStore.DefaultPropertyName;

    private const int DefaultTimeoutMs = 30_000;
    private const int DefaultMaxRedirects = 5;

    private readonly IHttpClientFactory _http;
    private readonly CredentialTypeCatalog _credentialTypes;
    private readonly EgressGuard _egress;
    private readonly NodeRequestBudget _quota;
    private readonly IBinaryStore _binary;

    public HttpRequestNodeExecutor(
        IHttpClientFactory http,
        CredentialTypeCatalog credentialTypes,
        EgressGuard egress,
        NodeRequestBudget quota,
        IBinaryStore binary)
    {
        _http = http;
        _credentialTypes = credentialTypes;
        _egress = egress;
        _quota = quota;
        _binary = binary;
    }

    public string Type => TypeName;

    public async Task<NodeExecutionOutcome> RunAsync(NodeExecutionContext context, CancellationToken ct)
    {
        var methodName = NodeValueAccess.ReadString(context.Parameters, "method") ?? "GET";
        var url = NodeValueAccess.ReadString(context.Parameters, "url");
        if (string.IsNullOrWhiteSpace(url))
            return NodeExecutionOutcome.Failed("url is required", "invalid_parameter");
        if (!Uri.TryCreate(url, UriKind.Absolute, out var uri))
            return NodeExecutionOutcome.Failed($"'{url}' is not a valid absolute URL", "invalid_url");

        var options = context.Parameters["options"] as JsonObject ?? new JsonObject();
        var timeoutMs = NodeValueAccess.ReadInt(options, "timeoutMs", DefaultTimeoutMs);
        if (timeoutMs <= 0)
            timeoutMs = DefaultTimeoutMs;
        var neverError = NodeValueAccess.ReadBool(options, "neverError");
        var responseFormat = NodeValueAccess.ReadString(options, "responseFormat") ?? "autodetect";
        var putOutputInField = NodeValueAccess.ReadString(options, "putOutputInField");
        var outputPath = NodeValueAccess.ReadString(options, "outputPath");
        var followRedirects = NodeValueAccess.ReadBool(options, "followRedirects", true);
        // The operator policy is a ceiling: a node may ask for fewer, never more.
        var maxRedirects = Math.Clamp(
            NodeValueAccess.ReadInt(options, "maxRedirects", DefaultMaxRedirects), 0, _egress.MaxRedirects);

        HttpPaginationPlan? pagination;
        try
        {
            pagination = HttpPaginationPlan.From(options);
        }
        catch (InvalidOperationException ex)
        {
            return NodeExecutionOutcome.Failed(ex.Message, "invalid_pagination");
        }

        // Pagination only makes sense when a page can be parsed into items.
        if (pagination != null && responseFormat is "text" or "file")
            pagination = null;

        // Resolve the auth intent once; the credential itself is injected per hop
        // and dropped when a redirect crosses to another host.
        var authentication = NodeValueAccess.ReadString(context.Parameters, "authentication") ?? "none";
        var hasCredential = context.Credentials.TryGetValue(CredentialAlias, out var credential);
        if (string.Equals(authentication, "credential", StringComparison.Ordinal) && !hasCredential)
            return NodeExecutionOutcome.Failed(
                $"no credential is configured for alias '{CredentialAlias}'", "missing_credential");

        using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
        timeout.CancelAfter(timeoutMs);

        var client = _http.CreateClient(
            NodeValueAccess.ReadBool(options, "ignoreSslIssues") ? InsecureClientName : Type);

        var method = new HttpMethod(methodName.ToUpperInvariant());
        var sendBody = NodeValueAccess.ReadBool(context.Parameters, "sendBody");
        var body = context.Parameters["body"]?.DeepClone();
        var sendCredential = !string.Equals(authentication, "none", StringComparison.Ordinal);
        var queryOverrides = new Dictionary<string, string>(StringComparer.Ordinal);
        var headerOverrides = new Dictionary<string, string>(StringComparer.Ordinal);

        // pageNumber mode owns its parameter from the first request on, so the
        // starting value is deterministic regardless of what the base URL carries.
        if (pagination?.InitialStep() is { } initial)
            body = ApplyStep(initial, pagination, body, queryOverrides, headerOverrides);

        var items = new List<FlowItem>();
        var currentUri = uri;
        var page = 0;
        var pageNumber = pagination?.PageStart ?? 1;

        while (true)
        {
            page++;
            var pageUri = ApplyQuery(currentUri, queryOverrides);

            var pageResult = await SendPageAsync(
                context, client, timeout, timeoutMs, pageUri, method, sendBody, body,
                headerOverrides, sendCredential, hasCredential, credential,
                followRedirects, maxRedirects, ct);

            if (pageResult.Failure != null)
                return pageResult.Failure;

            var status = pageResult.Status;
            if (status >= 400 && !neverError)
            {
                return new NodeExecutionOutcome
                {
                    Outputs = Array.Empty<IReadOnlyList<FlowItem>>(),
                    Failure = new NodeFailure(
                        $"HTTP {status}", "http_error", NodeValueAccess.Truncate(pageResult.Body), status),
                };
            }

            // A configured stop status ends pagination without contributing the
            // (normally empty) stop page to the output.
            if (pagination != null && pagination.IsStopStatus(status))
                break;

            var build = await BuildItemsAsync(
                pageResult, responseFormat, putOutputInField, outputPath, timeout.Token);
            if (build.Error != null)
                return NodeExecutionOutcome.Failed(build.Error, "invalid_response");

            var pageItems = build.Items;

            var itemLimitReached = AddItems(items, pageItems, pagination?.MaxItems ?? 0);

            if (pagination == null)
                break;
            if (itemLimitReached)
                break;
            if (pagination.IsStopPage(pageItems.Count))
                break;
            if (pagination.IsPageLimitReached(page))
                break;

            var step = pagination.TryNext(pageResult.Parsed, ref pageNumber);
            if (step == null)
                break;

            if (step.NextUrl != null)
            {
                if (!TryResolveNext(step.NextUrl, currentUri, out var nextUri))
                {
                    return NodeExecutionOutcome.Failed(
                        $"invalid pagination next URL '{step.NextUrl}'", "invalid_redirect");
                }
                currentUri = nextUri;
                // The provider's next URL already carries the query; do not
                // re-apply this run's overrides on top of it.
                queryOverrides.Clear();
            }
            else
            {
                body = ApplyStep(step, pagination, body, queryOverrides, headerOverrides);
            }
        }

        return NodeExecutionOutcome.Single(items);
    }
}