feat(delivery): apply JMESPath payload transformation before delivery (ADR-003 Phase 2)

DeliveryWorker now consults the new IPayloadTransformer abstraction (backed by JmesPath.Net 1.1.0) before signing and POSTing the payload. The transformed payload is what gets signed, so the receiver verifies the signature against the body they actually receive.

The delivery pipeline is fail-safe by construction. Transformation only runs when (a) the global TransformationOptions.Enabled flag is true, (b) the endpoint has TransformEnabled set, and (c) TransformExpression is non-empty. Any failure — invalid expression, 100 ms timeout, output exceeding 256 KB UTF-8, or unparseable JSON — returns PayloadTransformResult.FailOpen, the worker logs a warning, and delivery proceeds with the original payload. Two new metrics expose the split: webhookengine.transformations.applied counts successful applications, and webhookengine.transformations.failed_open counts fallbacks.

JmesPathPayloadTransformer wraps the library call in Task.Run + Task.Wait(timeout) so a runaway expression cannot stall the delivery thread; output size is checked via Encoding.UTF8.GetByteCount before returning success. The class is stateless and registered as a singleton.

The configuration surface lives under WebhookEngine:Transformation in appsettings (Enabled=true, TimeoutMs=100, MaxOutputBytes=262144). Six unit tests under WebhookEngine.Infrastructure.Tests.Services cover the reshape happy path, identity selector, invalid syntax, empty expression, oversized output, and invalid JSON cases. README test count bumped from 136 to 142.

Author: voyvodka

CHANGELOG.md

@@ -8,7 +8,8 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht
 ## [Unreleased]
 
 ### Added
-- **Payload transformation schema and API (ADR-003 Phase 1):** endpoints now accept `transformExpression` (JMESPath, max 4096 chars), `transformEnabled` (kill switch, default `false`), and a server-managed `transformValidatedAt` timestamp on create/update. Both the public Bearer-key API (`POST /api/v1/endpoints`, `PUT /api/v1/endpoints/{id}`) and the dashboard endpoints (`POST /api/v1/dashboard/endpoints`, `PUT /api/v1/dashboard/endpoints/{id}`) carry the new fields, and `EndpointResponseDto` exposes them on read. Stored only — pipeline integration (delivery-time application with `JmesPath.Net` + 100 ms timeout + fail-open) and the dashboard editor land in ADR-003 Phase 2 and Phase 3 respectively.
+- **Payload transformation delivery integration (ADR-003 Phase 2):** the `DeliveryWorker` now applies the per-endpoint JMESPath expression to the payload before signing and POSTing. Backed by the new `IPayloadTransformer` abstraction and `JmesPathPayloadTransformer` (JmesPath.Net 1.1.0). Hard guardrails enforced at delivery time: 100 ms wall-clock timeout, 256 KB output cap, and a global kill switch via `WebhookEngine:Transformation:Enabled` (defaults to `true`). Every transformation is fail-open — invalid expressions, timeouts, oversized output, or invalid JSON fall back to the original payload with a warning log. New OpenTelemetry counters `webhookengine.transformations.applied` and `webhookengine.transformations.failed_open` track success vs fallback. Six unit tests cover identity, reshape, invalid expression, empty expression, output-size, and invalid-json paths.
+- **Payload transformation schema and API (ADR-003 Phase 1):** endpoints now accept `transformExpression` (JMESPath, max 4096 chars), `transformEnabled` (kill switch, default `false`), and a server-managed `transformValidatedAt` timestamp on create/update. Both the public Bearer-key API (`POST /api/v1/endpoints`, `PUT /api/v1/endpoints/{id}`) and the dashboard endpoints (`POST /api/v1/dashboard/endpoints`, `PUT /api/v1/dashboard/endpoints/{id}`) carry the new fields, and `EndpointResponseDto` exposes them on read. The dashboard expression editor and live preview land in ADR-003 Phase 3.
 - **Security automations:** CodeQL workflow (csharp + javascript-typescript, push/PR/Mondays at 06:30 UTC), Dependency Review action on PRs (high-severity fail + GPL/LGPL/AGPL/EUPL/SSPL deny-list), and Dependabot config covering NuGet, npm, GitHub Actions, and Docker base images. Five repo labels (`dependencies`, `nuget`, `npm`, `ci`, `docker`) created to support the Dependabot config.
 
 ### Changed

README.md

@@ -272,7 +272,7 @@ All configuration is via `appsettings.json` or environment variables (double-und
 # Build
 dotnet build WebhookEngine.sln
 
-# Run all tests (136 tests)
+# Run all tests (142 tests)
 dotnet test WebhookEngine.sln
 
 # Run specific test project

src/WebhookEngine.API/Program.cs

@@ -36,6 +36,7 @@
 builder.Services.Configure<DashboardAuthOptions>(builder.Configuration.GetSection(DashboardAuthOptions.SectionName));
 builder.Services.Configure<RetentionOptions>(builder.Configuration.GetSection(RetentionOptions.SectionName));
 builder.Services.Configure<RateLimitOptions>(builder.Configuration.GetSection(RateLimitOptions.SectionName));
+builder.Services.Configure<TransformationOptions>(builder.Configuration.GetSection(TransformationOptions.SectionName));
 
 // Database
 builder.Services.AddDbContext<WebhookDbContext>(options =>
@@ -69,6 +70,7 @@
 builder.Services.AddSingleton<IMessageStateMachine, MessageStateMachine>();
 builder.Services.AddSingleton<IDeliveryNotifier, SignalRDeliveryNotifier>();
 builder.Services.AddSingleton<IDevTrafficGenerator, DevTrafficGenerator>();
+builder.Services.AddSingleton<IPayloadTransformer, JmesPathPayloadTransformer>();
 
 // Background Workers
 builder.Services.AddHostedService<DeliveryWorker>();

src/WebhookEngine.Core/Interfaces/IPayloadTransformer.cs

@@ -0,0 +1,24 @@
+namespace WebhookEngine.Core.Interfaces;
+
+/// <summary>
+/// Applies a declarative payload transformation (JMESPath expression) to a JSON payload
+/// before delivery. Implementations must be fail-safe: any error during transformation
+/// returns <see cref="PayloadTransformResult.FailOpen"/> so the caller can fall back to
+/// the original payload, never blocking delivery (ADR-003).
+/// </summary>
+public interface IPayloadTransformer
+{
+    /// <summary>
+    /// Apply <paramref name="expression"/> to <paramref name="payload"/> with built-in
+    /// timeout and output-size guards. Always returns a result; check
+    /// <see cref="PayloadTransformResult.IsSuccess"/> to decide whether to use the
+    /// transformed payload or fall back to the original.
+    /// </summary>
+    PayloadTransformResult Transform(string expression, string payload);
+}
+
+public sealed record PayloadTransformResult(bool IsSuccess, string? TransformedPayload, string? Error)
+{
+    public static PayloadTransformResult Success(string transformed) => new(true, transformed, null);
+    public static PayloadTransformResult FailOpen(string error) => new(false, null, error);
+}

src/WebhookEngine.Core/Metrics/WebhookMetrics.cs

@@ -21,6 +21,8 @@ public sealed class WebhookMetrics
     private readonly Counter<long> _staleLockRecovered;
     private readonly Histogram<double> _deliveryDurationMs;
     private readonly UpDownCounter<long> _queueDepth;
+    private readonly Counter<long> _transformationsApplied;
+    private readonly Counter<long> _transformationsFailedOpen;
 
     public WebhookMetrics(IMeterFactory meterFactory)
     {
@@ -80,6 +82,16 @@ public WebhookMetrics(IMeterFactory meterFactory)
             "webhookengine.queue.depth",
             unit: "{message}",
             description: "Approximate queue depth (enqueue increments, dequeue decrements)");
+
+        _transformationsApplied = meter.CreateCounter<long>(
+            "webhookengine.transformations.applied",
+            unit: "{transformation}",
+            description: "Successful payload transformations applied before delivery");
+
+        _transformationsFailedOpen = meter.CreateCounter<long>(
+            "webhookengine.transformations.failed_open",
+            unit: "{transformation}",
+            description: "Payload transformations that failed and fell back to the original payload");
     }
 
     public void RecordMessageEnqueued(int count = 1) => _messagesEnqueued.Add(count);
@@ -111,4 +123,8 @@ public void RecordDeliveryFailure(double durationMs)
     public void RecordQueueEnqueue(int count = 1) => _queueDepth.Add(count);
 
     public void RecordQueueDequeue(int count) => _queueDepth.Add(-count);
+
+    public void RecordTransformationApplied() => _transformationsApplied.Add(1);
+
+    public void RecordTransformationFailedOpen() => _transformationsFailedOpen.Add(1);
 }

src/WebhookEngine.Core/Options/TransformationOptions.cs

@@ -0,0 +1,31 @@
+namespace WebhookEngine.Core.Options;
+
+/// <summary>
+/// Global configuration for the payload transformation pipeline (ADR-003).
+/// All limits are enforced at delivery time before the HTTP POST. Per-endpoint
+/// transformation toggles live on the <c>Endpoint</c> entity (TransformEnabled,
+/// TransformExpression).
+/// </summary>
+public class TransformationOptions
+{
+    public const string SectionName = "WebhookEngine:Transformation";
+
+    /// <summary>
+    /// Global kill switch. When false, no transformations are applied regardless
+    /// of per-endpoint settings — every delivery uses the original payload.
+    /// </summary>
+    public bool Enabled { get; set; } = true;
+
+    /// <summary>
+    /// Hard timeout per JMESPath evaluation in milliseconds. Pathological
+    /// expressions are aborted and the delivery falls back to the original payload.
+    /// </summary>
+    public int TimeoutMs { get; set; } = 100;
+
+    /// <summary>
+    /// Maximum size of the transformed payload in bytes (UTF-8). Prevents
+    /// transformations from inflating output beyond the message-size budget.
+    /// Matches the input payload limit (256 KB).
+    /// </summary>
+    public int MaxOutputBytes { get; set; } = 256 * 1024;
+}

src/WebhookEngine.Infrastructure/Services/JmesPathPayloadTransformer.cs

@@ -0,0 +1,86 @@
+using System.Text;
+using DevLab.JmesPath;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using WebhookEngine.Core.Interfaces;
+using WebhookEngine.Core.Options;
+
+namespace WebhookEngine.Infrastructure.Services;
+
+/// <summary>
+/// JMESPath-backed payload transformer (ADR-003). Wraps the JmesPath.Net library
+/// with a hard timeout, an output-size guard, and a fail-open contract: any
+/// error returns <see cref="PayloadTransformResult.FailOpen"/> so callers fall
+/// back to the original payload. Stateless and safe to register as a singleton.
+/// </summary>
+public sealed class JmesPathPayloadTransformer : IPayloadTransformer
+{
+    private readonly TransformationOptions _options;
+    private readonly ILogger<JmesPathPayloadTransformer> _logger;
+    private readonly JmesPath _jmesPath = new();
+
+    public JmesPathPayloadTransformer(
+        IOptions<TransformationOptions> options,
+        ILogger<JmesPathPayloadTransformer> logger)
+    {
+        _options = options.Value;
+        _logger = logger;
+    }
+
+    public PayloadTransformResult Transform(string expression, string payload)
+    {
+        if (string.IsNullOrWhiteSpace(expression))
+        {
+            return PayloadTransformResult.FailOpen("Expression is empty.");
+        }
+
+        // Run the JMESPath evaluation on a thread-pool task so we can enforce a
+        // hard wall-clock timeout even if the expression hits a pathological
+        // pattern that the parser does not detect ahead of time.
+        var task = Task.Run(() => _jmesPath.Transform(payload, expression));
+
+        try
+        {
+            if (!task.Wait(TimeSpan.FromMilliseconds(_options.TimeoutMs)))
+            {
+                _logger.LogWarning(
+                    "JMESPath transformation timed out after {TimeoutMs}ms for expression {Expression}",
+                    _options.TimeoutMs, expression);
+                return PayloadTransformResult.FailOpen($"Timeout after {_options.TimeoutMs}ms.");
+            }
+        }
+        catch (AggregateException ex) when (ex.InnerException is not null)
+        {
+            _logger.LogWarning(
+                ex.InnerException,
+                "JMESPath transformation failed for expression {Expression}",
+                expression);
+            return PayloadTransformResult.FailOpen(ex.InnerException.Message);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "JMESPath transformation failed for expression {Expression}", expression);
+            return PayloadTransformResult.FailOpen(ex.Message);
+        }
+
+        var transformed = task.Result;
+        if (transformed is null)
+        {
+            return PayloadTransformResult.FailOpen("JMESPath returned null result.");
+        }
+
+        // Reject results that exceed the output budget to keep delivery payloads
+        // bounded. UTF-8 byte length matches what the HTTP body will weigh.
+        var byteCount = Encoding.UTF8.GetByteCount(transformed);
+        if (byteCount > _options.MaxOutputBytes)
+        {
+            _logger.LogWarning(
+                "JMESPath transformation output exceeded {MaxOutputBytes} bytes ({ActualBytes}) for expression {Expression}",
+                _options.MaxOutputBytes, byteCount, expression);
+            return PayloadTransformResult.FailOpen(
+                $"Output size {byteCount} exceeds limit {_options.MaxOutputBytes}.");
+        }
+
+        return PayloadTransformResult.Success(transformed);
+    }
+}

src/WebhookEngine.Infrastructure/WebhookEngine.Infrastructure.csproj

@@ -7,6 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="JmesPath.Net" Version="1.1.0" />
     <PackageReference Include="Microsoft.EntityFrameworkCore" Version="10.0.5" />
     <PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="10.0.1" />
     <PackageReference Include="Microsoft.Extensions.Http" Version="10.0.5" />

src/WebhookEngine.Worker/DeliveryWorker.cs

@@ -19,7 +19,9 @@ public class DeliveryWorker : BackgroundService
     private readonly ILogger<DeliveryWorker> _logger;
     private readonly DeliveryOptions _deliveryOptions;
     private readonly RetryPolicyOptions _retryPolicy;
+    private readonly TransformationOptions _transformationOptions;
     private readonly IEndpointRateLimiter _endpointRateLimiter;
+    private readonly IPayloadTransformer _payloadTransformer;
     private readonly WebhookMetrics? _metrics;
     private readonly string _workerId = $"worker_{Environment.MachineName}_{Guid.NewGuid().ToString("N")[..8]}";
 
@@ -28,14 +30,18 @@ public DeliveryWorker(
         ILogger<DeliveryWorker> logger,
         IOptions<DeliveryOptions> deliveryOptions,
         IOptions<RetryPolicyOptions> retryPolicy,
+        IOptions<TransformationOptions> transformationOptions,
         IEndpointRateLimiter endpointRateLimiter,
+        IPayloadTransformer payloadTransformer,
         WebhookMetrics? metrics = null)
     {
         _serviceProvider = serviceProvider;
         _logger = logger;
         _deliveryOptions = deliveryOptions.Value;
         _retryPolicy = retryPolicy.Value;
+        _transformationOptions = transformationOptions.Value;
         _endpointRateLimiter = endpointRateLimiter;
+        _payloadTransformer = payloadTransformer;
         _metrics = metrics;
     }
 
@@ -169,9 +175,15 @@ private async Task ProcessMessageAsync(
                 return;
             }
 
-            // Sign the payload
+            // Apply payload transformation (ADR-003) before signing — the receiver
+            // verifies the signature against the body they actually receive, so the
+            // transformed payload is what we sign. Fail-open: any error keeps the
+            // original payload and lets delivery proceed.
+            var deliveryPayload = ApplyTransformation(message.Payload, endpoint, message.Id);
+
+            // Sign the (possibly transformed) payload
             var timestamp = DateTimeOffset.UtcNow.ToUnixTimeSeconds();
-            var signedHeaders = signingService.Sign(message.Id.ToString(), timestamp, message.Payload, signingSecret);
+            var signedHeaders = signingService.Sign(message.Id.ToString(), timestamp, deliveryPayload, signingSecret);
 
             var customHeaders = ParseCustomHeaders(endpoint.CustomHeadersJson);
             var requestHeaders = BuildRequestHeaders(signedHeaders, customHeaders);
@@ -181,7 +193,7 @@ private async Task ProcessMessageAsync(
             {
                 MessageId = message.Id.ToString(),
                 EndpointUrl = endpoint.Url,
-                Payload = message.Payload,
+                Payload = deliveryPayload,
                 SignedHeaders = signedHeaders,
                 CustomHeaders = customHeaders
             };
@@ -280,6 +292,34 @@ private DateTime CalculateNextRetryAt(int currentAttempt)
         return DateTime.UtcNow.AddSeconds(backoffSeconds);
     }
 
+    private string ApplyTransformation(string originalPayload, Core.Entities.Endpoint endpoint, Guid messageId)
+    {
+        // Skip when globally disabled or per-endpoint not configured.
+        if (!_transformationOptions.Enabled
+            || !endpoint.TransformEnabled
+            || string.IsNullOrWhiteSpace(endpoint.TransformExpression))
+        {
+            return originalPayload;
+        }
+
+        var result = _payloadTransformer.Transform(endpoint.TransformExpression, originalPayload);
+
+        if (result.IsSuccess && result.TransformedPayload is not null)
+        {
+            _metrics?.RecordTransformationApplied();
+            _logger.LogInformation(
+                "Applied JMESPath transformation for message {MessageId} on endpoint {EndpointId}",
+                messageId, endpoint.Id);
+            return result.TransformedPayload;
+        }
+
+        _metrics?.RecordTransformationFailedOpen();
+        _logger.LogWarning(
+            "JMESPath transformation failed for message {MessageId} on endpoint {EndpointId}; falling back to original payload. Error: {Error}",
+            messageId, endpoint.Id, result.Error);
+        return originalPayload;
+    }
+
     private static Dictionary<string, string> ParseCustomHeaders(string? customHeadersJson)
     {
         if (string.IsNullOrWhiteSpace(customHeadersJson))