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

using System.Text.Json.Nodes;
using w4c_workflows.Models.Nodes;

namespace w4c_workflows.Services.Nodes.Executors;

/// <summary>
/// Aggregate node. Two modes:
///   - <c>allItemData</c>: folds the batch into a single item whose destination
///     field holds the array of every item payload;
///   - <c>individualFields</c>: groups items by every field that is not being
///     aggregated and emits one item per group, carrying the grouping fields
///     plus the aggregated columns (count/sum/average/min/max).
/// With no fields selected the whole batch collapses to a single count.
/// Aggregated field names use dot notation for nested values.
/// </summary>
public sealed class AggregateNodeExecutor : INodeExecutor
{
    public string Type => "core.aggregate";

    public Task<NodeExecutionOutcome> RunAsync(NodeExecutionContext context, CancellationToken ct)
    {
        var items = context.Input(0);
        var mode = NodeValueAccess.ReadString(context.Parameters, "mode") ?? "individualFields";

        if (string.Equals(mode, "allItemData", StringComparison.OrdinalIgnoreCase))
        {
            var destination = NodeValueAccess.ReadString(context.Parameters, "destinationFieldName");
            if (string.IsNullOrWhiteSpace(destination))
                destination = "data";

            var data = new JsonArray();
            foreach (var item in items)
                data.Add(item.Json.DeepClone());

            return Task.FromResult(NodeExecutionOutcome.Single(new[]
            {
                new FlowItem { Json = new JsonObject { [destination] = data } },
            }));
        }

        var specs = new List<AggregationSpec>();
        foreach (var entry in NodeValueAccess.ReadEntries(context.Parameters, "fields"))
        {
            var field = NodeValueAccess.ReadString(entry, "fieldName");
            var aggregation = NodeValueAccess.ReadString(entry, "aggregation") ?? "count";
            if (!string.Equals(aggregation, "count", StringComparison.OrdinalIgnoreCase)
                && string.IsNullOrWhiteSpace(field))
            {
                return Task.FromResult(NodeExecutionOutcome.Failed(
                    $"aggregation '{aggregation}' needs a field name", "invalid_parameter"));
            }

            var rename = NodeValueAccess.ReadString(entry, "renameField");
            var outputName = !string.IsNullOrWhiteSpace(rename) ? rename!
                : !string.IsNullOrWhiteSpace(field) ? field!
                : "count";
            specs.Add(new AggregationSpec(field, aggregation, outputName));
        }

        if (specs.Count == 0)
        {
            return Task.FromResult(NodeExecutionOutcome.Single(new[]
            {
                new FlowItem { Json = new JsonObject { ["count"] = items.Count } },
            }));
        }

        // An aggregated field is consumed; every other field present on the items
        // forms the grouping key.
        var consumed = new HashSet<string>(StringComparer.Ordinal);
        foreach (var spec in specs)
        {
            if (!string.IsNullOrWhiteSpace(spec.FieldName))
                consumed.Add(TopLevel(spec.FieldName!));
        }

        var groupingKeys = new List<string>();
        var seenKey = new HashSet<string>(StringComparer.Ordinal);
        foreach (var item in items)
        {
            foreach (var (name, _) in item.Json)
            {
                if (!consumed.Contains(name) && seenKey.Add(name))
                    groupingKeys.Add(name);
            }
        }
        groupingKeys.Sort(StringComparer.Ordinal);

        var groups = new List<(JsonObject Key, List<FlowItem> Items)>();
        var indexByKey = new Dictionary<string, int>(StringComparer.Ordinal);
        foreach (var item in items)
        {
            var key = new JsonObject();
            foreach (var name in groupingKeys)
                key[name] = item.Json.TryGetPropertyValue(name, out var value) ? value?.DeepClone() : null;

            var canonical = NodeJsonCanonical.For(key);
            if (!indexByKey.TryGetValue(canonical, out var index))
            {
                indexByKey[canonical] = groups.Count;
                groups.Add((key, new List<FlowItem>()));
                index = groups.Count - 1;
            }
            groups[index].Items.Add(item);
        }

        var output = new List<FlowItem>(groups.Count);
        foreach (var (key, groupItems) in groups)
        {
            var json = (JsonObject)key.DeepClone();
            foreach (var spec in specs)
                json[spec.OutputName] = Aggregate(spec, groupItems);
            output.Add(new FlowItem { Json = json });
        }

        return Task.FromResult(NodeExecutionOutcome.Single(output));
    }

    private sealed record AggregationSpec(string? FieldName, string Aggregation, string OutputName);

    private static JsonNode? Aggregate(AggregationSpec spec, IReadOnlyList<FlowItem> items)
    {
        switch (spec.Aggregation.ToLowerInvariant())
        {
            case "count":
                return JsonValue.Create(items.Count);

            case "sum":
            {
                var sum = 0d;
                var any = false;
                foreach (var item in items)
                {
                    if (NodeValueAccess.TryReadNumber(Value(item, spec), out var number))
                    {
                        sum += number;
                        any = true;
                    }
                }
                return any ? JsonValue.Create(sum) : null;
            }

            case "average":
            {
                var sum = 0d;
                var count = 0;
                foreach (var item in items)
                {
                    if (NodeValueAccess.TryReadNumber(Value(item, spec), out var number))
                    {
                        sum += number;
                        count++;
                    }
                }
                return count > 0 ? JsonValue.Create(sum / count) : null;
            }

            case "min":
                return Extreme(spec, items, minimum: true);

            case "max":
                return Extreme(spec, items, minimum: false);

            default:
                return null;
        }
    }

    /// <summary>Numeric extreme when any value is numeric, otherwise an ordinal text extreme.</summary>
    private static JsonNode? Extreme(AggregationSpec spec, IReadOnlyList<FlowItem> items, bool minimum)
    {
        double? number = null;
        string? text = null;

        foreach (var item in items)
        {
            var value = Value(item, spec);
            if (value is null)
                continue;

            if (NodeValueAccess.TryReadNumber(value, out var candidate))
            {
                if (number is null || (minimum ? candidate < number : candidate > number))
                    number = candidate;
                continue;
            }

            var candidateText = value is JsonValue jsonValue
                && jsonValue.TryGetValue<string>(out var textValue)
                    ? textValue
                    : value.ToJsonString();
            if (text is null || (minimum
                    ? string.CompareOrdinal(candidateText, text) < 0
                    : string.CompareOrdinal(candidateText, text) > 0))
            {
                text = candidateText;
            }
        }

        if (number is not null)
            return JsonValue.Create(number.Value);
        return text is null ? null : JsonValue.Create(text);
    }

    private static JsonNode? Value(FlowItem item, AggregationSpec spec)
        => NodeJsonPath.Read(item.Json, spec.FieldName, dotNotation: true);

    private static string TopLevel(string field)
        => field.Split('.')[0];
}