Dapr Bindings & Aspire

dotnet new aspire-starter --output dapr-bindings-aspire

dotnet add package CommunityToolkit.Aspire.Hosting.Dapr

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: input-binding-kafka
spec:
  type: bindings.kafka
  version: v1
  metadata:
    - name: brokers
      value: localhost:19092
    - name: topics
      value: weather-input
    - name: consumerGroup
      value: weather-api
    - name: authRequired
      value: "false"
    - name: initialOffset
      value: newest
    - name: direction
      value: input

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: input-binding-rabbitmq
spec:
  type: bindings.rabbitmq
  version: v1
  metadata:
    - name: queueName
      value: weather-input-queue
    - name: host
      value: amqp://guest:guest@localhost:5672
    - name: durable
      value: "true"
    - name: deleteWhenUnused
      value: "false"
    - name: prefetchCount
      value: "0"
    - name: direction
      value: input

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: output-binding-kafka
spec:
  type: bindings.kafka
  version: v1
  metadata:
    - name: brokers
      value: localhost:19092
    - name: publishTopic
      value: weather-output
    - name: authRequired
      value: "false"
    - name: direction
      value: output

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: output-binding-rabbitmq
spec:
  type: bindings.rabbitmq
  version: v1
  metadata:
    - name: queueName
      value: weather-output-queue
    - name: host
      value: amqp://guest:guest@localhost:5672
    - name: durable
      value: "true"
    - name: deleteWhenUnused
      value: "false"
    - name: direction
      value: output

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: output-binding-redis
spec:
  type: bindings.redis
  version: v1
  metadata:
    - name: redisHost
      value: localhost:6379
    - name: redisPassword
      value: ""
    - name: enableTLS
      value: "false"
    - name: direction
      value: output

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: trigger-binding-kafka
spec:
  type: bindings.kafka
  version: v1
  metadata:
    - name: brokers
      value: localhost:19092
    - name: publishTopic
      value: weather-input
    - name: authRequired
      value: "false"
    - name: direction
      value: output

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: trigger-binding-rabbitmq
spec:
  type: bindings.rabbitmq
  version: v1
  metadata:
    - name: queueName
      value: weather-input-queue
    - name: host
      value: amqp://guest:guest@localhost:5672
    - name: durable
      value: "true"
    - name: deleteWhenUnused
      value: "false"
    - name: direction
      value: output

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: verify-output-binding-kafka
spec:
  type: bindings.kafka
  version: v1
  metadata:
    - name: brokers
      value: localhost:19092
    - name: topics
      value: weather-output
    - name: consumerGroup
      value: weather-output-verifier
    - name: authRequired
      value: "false"
    - name: initialOffset
      value: newest
    - name: direction
      value: input

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: verify-output-binding-rabbitmq
spec:
  type: bindings.rabbitmq
  version: v1
  metadata:
    - name: queueName
      value: weather-output-queue
    - name: host
      value: amqp://guest:guest@localhost:5672
    - name: durable
      value: "true"
    - name: deleteWhenUnused
      value: "false"
    - name: prefetchCount
      value: "0"
    - name: direction
      value: input

var builder = DistributedApplication.CreateBuilder(args);
 
var callbackToken = builder.AddParameter("weather-callback-token",
    new GenerateParameterDefault { MinLength = 32 }, secret: true);
 
// Dapr components path
var daprComponentsPath = Path.Combine(builder.AppHostDirectory, "components");
 
// Dashboard link only: the external rabbitmq:4-management container owns ports 5672 and 15672.
// This does not start a broker, create a Management UI, or bind either port.
builder.AddExternalService("rabbitmq-management", "http://localhost:15672");
 
// Console only; the external broker owns Kafka ports 19092 (Dapr) and 29092 (containers).
builder.AddContainer("redpanda-console", "docker.redpanda.com/redpandadata/console", "v3.10.0")
    .WithEndpoint(targetPort: 8080, scheme: "http", name: "http", isExternal: true)
    .WithEnvironment("KAFKA_BROKERS", "host.docker.internal:29092")
    .WithUrlForEndpoint("http", url =>
    {
        url.Url = "/";
        url.DisplayText = "Redpanda Console";
    });
 
// RedisInsight UI
builder.AddContainer("redisinsight", "redis/redisinsight", "latest")
    .WithEndpoint(targetPort: 5540, scheme: "http", name: "http", isExternal: true)
    .WithEnvironment("RI_APP_HOST", "0.0.0.0")
    .WithEnvironment("RI_APP_PORT", "5540")
    .WithEnvironment("RI_REDIS_HOST", "host.docker.internal")
    .WithEnvironment("RI_REDIS_PORT", "6379")
    .WithEnvironment("RI_REDIS_ALIAS", "Dapr binding Redis")
    .WithUrlForEndpoint("http", url =>
    {
        url.Url = "/";
        url.DisplayText = "Redis Insight";
    });
 
// ApiService project with Dapr sidecar
var apiService = builder.AddProject<Projects.dapr_bindings_aspire_ApiService>("apiservice")
    .WithEnvironment("WeatherFlow__CallbackToken", callbackToken)
    .WithHttpHealthCheck("/health")
    .WithDaprSidecar(new CommunityToolkit.Aspire.Hosting.Dapr.DaprSidecarOptions
    {
        ResourcesPaths = [daprComponentsPath]
    })
    .WithUrlForEndpoint("http", url =>
    {
        url.Url = "/";
        url.DisplayText = "API Service";
    })
    .WithUrlForEndpoint("https", url =>
    {
        url.Url = "/";
        url.DisplayText = "API Service";
    });
 
// Web App project
var webFrontend = builder.AddProject<Projects.dapr_bindings_aspire_Web>("webfrontend")
    .WithEnvironment("WeatherFlow__CallbackToken", callbackToken)
    .WithExternalHttpEndpoints()
    .WithHttpHealthCheck("/health")
    .WithReference(apiService)
    .WaitFor(apiService)
    .WithUrlForEndpoint("http", url =>
    {
        url.Url = "/";
        url.DisplayText = "Web App";
    })
    .WithUrlForEndpoint("https", url =>
    {
        url.Url = "/";
        url.DisplayText = "Web App";
    });
 
// Callbacks need reverse discovery
apiService.WithReference(webFrontend);
 
builder.Build().Run();

  docker run -d \
    --name dapr-rabbitmq \
    --hostname dapr-rabbitmq \
    -p 5672:5672 \
    -p 15672:15672 \
    rabbitmq:4-management

docker run -d \
    --name dapr-redpanda \
    --hostname dapr-redpanda \
    -p 19092:19092 \
    -p 29092:29092 \
    docker.redpanda.com/redpandadata/redpanda:v26.2.2 \
    redpanda start \
    --mode dev-container \
    --smp 1 \
    --memory 512M \
    --node-id 0 \
    --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092,containers://0.0.0.0:29092 \
    --advertise-kafka-addr internal://dapr-redpanda:9092,external://localhost:19092,containers://host.docker.internal:29092

docker ps --filter name=dapr-r

CONTAINER ID   IMAGE                                               COMMAND                  CREATED          STATUS          PORTS                                                                                              NAMES
9d0361b75952   rabbitmq:4-management                               "docker-entrypoint.s…"   42 minutes ago   Up 42 minutes   0.0.0.0:5672->5672/tcp, [::]:5672->5672/tcp, 0.0.0.0:15672->15672/tcp, [::]:15672->15672/tcp       dapr-rabbitmq
5a123fd3764c   docker.redpanda.com/redpandadata/redpanda:v26.2.2   "/entrypoint.sh redp…"   43 minutes ago   Up 43 minutes   0.0.0.0:19092->19092/tcp, [::]:19092->19092/tcp, 0.0.0.0:29092->29092/tcp, [::]:29092->29092/tcp   dapr-redpanda

namespace dapr_bindings_aspire.Contracts;
 
public sealed record WeatherForecastTriggerRequest(
    int? Days,
    DateOnly? StartDate,
    string? CorrelationId,
    string? OutputBinding,
    string? CallbackUrl);
 
public sealed record WeatherBindingOption(
    string Id,
    string Label,
    string BindingName);
 
public sealed record WeatherFlowOptions(
    WeatherBindingOption[] InputBindings,
    WeatherBindingOption[] OutputBindings);
 
public sealed record WeatherFlowRequest(
    string InputBinding,
    string OutputBinding,
    string? CallbackUrl,
    string? CorrelationId = null);
 
public sealed record WeatherFlowStartedResponse(
    bool Started,
    string Message,
    string CorrelationId,
    string InputBinding,
    string InputLabel,
    string OutputBinding,
    string OutputLabel);
 
public sealed record WeatherFlowNotification(
    string CorrelationId,
    string Status,
    string Message,
    WeatherForecastResponse Forecast);
 
public sealed record WeatherForecast(DateOnly Date, int TemperatureC, string? Summary)
{
    public int TemperatureF => 32 + (int)(TemperatureC / 0.5556);
}
 
public sealed record WeatherForecastResponse(
    DateTimeOffset? GeneratedAt,
    WeatherForecast[] Forecasts,
    string? CorrelationId = null,
    string? InputBinding = null,
    string? InputLabel = null,
    string? OutputBinding = null,
    string? OutputLabel = null,
    string? OutputStatus = null);
 
public static class FlowStatus
{
    public const string InputReceived = "input-received";
    public const string ForecastGenerated = "forecast-generated";
    public const string OutputAccepted = "output-accepted";
    public const string Verified = "verified";
    public const string Failed = "failed";
}

using dapr_bindings_aspire.ApiService.WeatherForecasting;
using Microsoft.AspNetCore.Http.HttpResults;
 
var builder = WebApplication.CreateBuilder(args);
 
builder.AddServiceDefaults();
 
builder.Services.AddProblemDetails();
 
builder.Services.AddOpenApi();
 
builder.Services.AddHttpClient();
builder.Services.AddHttpClient("weather-notifications")
    .ConfigurePrimaryHttpMessageHandler(() => new HttpClientHandler { AllowAutoRedirect = false });
builder.Services.AddSingleton(TimeProvider.System);
builder.Services.AddSingleton<InMemoryWeatherFlowStore>();
builder.Services.AddSingleton<DaprBindingClient>();
builder.Services.AddSingleton<WeatherFlowNotificationDispatcher>();
builder.Services.AddHostedService(services => services.GetRequiredService<WeatherFlowNotificationDispatcher>());
builder.Services.AddSingleton<IWeatherFlowCoordinator, WeatherFlowCoordinator>();
 
var app = builder.Build();
 
app.UseExceptionHandler();
 
if (app.Environment.IsDevelopment())
{
    app.MapOpenApi();
}
 
app.MapGet("/", () => "API service is running. Open the Web App to use the bindings demo.");
 
app.MapGet("/weatherforecast/flows/options", (IWeatherFlowCoordinator weatherFlowCoordinator) =>
    TypedResults.Ok(weatherFlowCoordinator.GetFlowOptions()))
.WithName("GetWeatherFlowOptions")
.WithSummary("Get selectable weather binding flow options.")
.WithDescription("Returns the input and output Dapr binding options exposed by the weather demo.")
.Produces<WeatherFlowOptions>(StatusCodes.Status200OK);
 
app.MapPost("/weatherforecast/flows/start", async Task<Results<Accepted<WeatherFlowStartedResponse>, ProblemHttpResult>> (
    WeatherFlowRequest request,
    IWeatherFlowCoordinator weatherFlowCoordinator,
    CancellationToken cancellationToken) =>
{
    try
    {
        var result = await weatherFlowCoordinator.StartFlowAsync(request, cancellationToken);
        return TypedResults.Accepted((string?)null, result);
    }
    catch (FlowValidationException ex)
    {
        return TypedResults.Problem(ex.Message, statusCode: StatusCodes.Status400BadRequest);
    }
    catch (FlowCapacityException ex)
    {
        return TypedResults.Problem(ex.Message, statusCode: StatusCodes.Status503ServiceUnavailable);
    }
    catch (HttpRequestException)
    {
        return TypedResults.Problem("The trigger binding is unavailable.", statusCode: StatusCodes.Status502BadGateway);
    }
    catch (Polly.CircuitBreaker.BrokenCircuitException)
    {
        return TypedResults.Problem("The trigger binding is temporarily unavailable.", statusCode: StatusCodes.Status503ServiceUnavailable);
    }
    catch (Polly.Timeout.TimeoutRejectedException)
    {
        return TypedResults.Problem("The trigger binding timed out.", statusCode: StatusCodes.Status504GatewayTimeout);
    }
    catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
    {
        return TypedResults.Problem("The trigger binding timed out.", statusCode: StatusCodes.Status504GatewayTimeout);
    }
})
.WithName("StartWeatherBindingFlow")
.WithSummary("Start a selected Dapr binding flow.")
.WithDescription("Publishes a trigger message to the selected input binding; the Dapr input binding later calls ApiService.")
.Produces<WeatherFlowStartedResponse>(StatusCodes.Status202Accepted)
.ProducesProblem(StatusCodes.Status400BadRequest)
.ProducesProblem(StatusCodes.Status502BadGateway)
.ProducesProblem(StatusCodes.Status503ServiceUnavailable)
.ProducesProblem(StatusCodes.Status504GatewayTimeout);
 
app.MapPost("/input-binding-rabbitmq", async Task<Ok<WeatherForecastResponse>> (
    HttpRequest request,
    IWeatherFlowCoordinator weatherFlowCoordinator,
    CancellationToken cancellationToken) =>
{
    using var reader = new StreamReader(request.Body);
    var body = await reader.ReadToEndAsync(cancellationToken);
    var forecast = await weatherFlowCoordinator.StoreInputBindingAsync("rabbitmq", body, cancellationToken);
    return TypedResults.Ok(forecast);
})
.WithName("ReceiveRabbitMqWeatherForecast")
.WithSummary("Receive a RabbitMQ input binding event.")
.WithDescription("Dapr calls this endpoint when a message arrives on the input-binding-rabbitmq RabbitMQ binding.")
.Produces<WeatherForecastResponse>(StatusCodes.Status200OK);
 
app.MapPost("/input-binding-kafka", async Task<Ok<WeatherForecastResponse>> (
    HttpRequest request,
    IWeatherFlowCoordinator weatherFlowCoordinator,
    CancellationToken cancellationToken) =>
{
    using var reader = new StreamReader(request.Body);
    var body = await reader.ReadToEndAsync(cancellationToken);
    var forecast = await weatherFlowCoordinator.StoreInputBindingAsync("kafka", body, cancellationToken);
    return TypedResults.Ok(forecast);
})
.WithName("ReceiveKafkaWeatherForecast")
.WithSummary("Receive a Kafka input binding event.")
.WithDescription("Dapr calls this endpoint when a message arrives on the input-binding-kafka Kafka binding.")
.Produces<WeatherForecastResponse>(StatusCodes.Status200OK);
 
app.MapPost("/verify-output-binding-rabbitmq", async Task<Ok> (
    HttpRequest request,
    IWeatherFlowCoordinator weatherFlowCoordinator,
    CancellationToken cancellationToken) =>
{
    using var reader = new StreamReader(request.Body);
    await weatherFlowCoordinator.VerifyOutputBindingAsync("rabbitmq", await reader.ReadToEndAsync(cancellationToken), cancellationToken);
    return TypedResults.Ok();
})
.WithName("VerifyRabbitMqWeatherOutput")
.WithSummary("Verify a RabbitMQ output binding delivery.");
 
app.MapPost("/verify-output-binding-kafka", async Task<Ok> (
    HttpRequest request,
    IWeatherFlowCoordinator weatherFlowCoordinator,
    CancellationToken cancellationToken) =>
{
    using var reader = new StreamReader(request.Body);
    await weatherFlowCoordinator.VerifyOutputBindingAsync("kafka", await reader.ReadToEndAsync(cancellationToken), cancellationToken);
    return TypedResults.Ok();
})
.WithName("VerifyKafkaWeatherOutput")
.WithSummary("Verify a Kafka output binding delivery.");
 
app.MapDefaultEndpoints();
 
app.Run();

global using dapr_bindings_aspire.Contracts;

using System.Net.Http.Json;
 
namespace dapr_bindings_aspire.ApiService.WeatherForecasting;
 
public sealed class DaprBindingClient(IHttpClientFactory clients, IConfiguration configuration)
{
    public Task<HttpResponseMessage> InvokeAsync(string bindingName, DaprBindingRequest request, CancellationToken cancellationToken)
    {
        var port = configuration["DAPR_HTTP_PORT"] ?? "3500";
        return clients.CreateClient("dapr").PostAsJsonAsync(
            $"http://127.0.0.1:{port}/v1.0/bindings/{bindingName}", request, cancellationToken);
    }
}
 
public sealed record DaprBindingRequest(string Operation, object? Data, Dictionary<string, string> Metadata);
 
public sealed class FlowValidationException(string message) : Exception(message);
public sealed class FlowCapacityException() : Exception("Too many retained flows. Try again later.");

namespace dapr_bindings_aspire.ApiService.WeatherForecasting;
 
// A bounded, single-process idempotency window. Entries are retained after completion
// so duplicate broker deliveries reuse the same task and generated forecast.
public sealed class InMemoryWeatherFlowStore(TimeProvider clock)
{
    private readonly object gate = new();
    private readonly Dictionary<string, Entry> entries = new(StringComparer.Ordinal);
    public const int Capacity = 1000;
    public static readonly TimeSpan Retention = TimeSpan.FromMinutes(10);
 
    public Entry Register(string id, string input, string output)
    {
        if (string.IsNullOrWhiteSpace(id) || id.Length > 128)
            throw new FlowValidationException("Correlation ID must contain 1–128 characters.");
        lock (gate)
        {
            RemoveExpired();
            if (entries.TryGetValue(id, out var existing))
            {
                if (existing.Input != input || existing.Output != output)
                    throw new FlowValidationException("Correlation ID is already assigned to another binding selection.");
                return existing;
            }
            if (entries.Count >= Capacity)
                throw new FlowCapacityException();
            var entry = new Entry(input, output, clock.GetUtcNow() + Retention);
            entries.Add(id, entry);
            return entry;
        }
    }
 
    public Entry? Find(string id)
    {
        lock (gate)
        {
            RemoveExpired();
            return entries.GetValueOrDefault(id);
        }
    }
 
    public Entry[] Snapshot()
    {
        lock (gate)
        {
            RemoveExpired();
            return entries.Values.ToArray();
        }
    }
 
    private void RemoveExpired()
    {
        foreach (var id in entries.Where(pair => pair.Value.ExpiresAt <= clock.GetUtcNow()).Select(pair => pair.Key).ToArray())
            entries.Remove(id);
    }
 
    public sealed class Entry(string input, string output, DateTimeOffset expiresAt)
    {
        public object Gate { get; } = new();
        public string Input { get; } = input;
        public string Output { get; } = output;
        public DateTimeOffset ExpiresAt { get; } = expiresAt;
        public Lazy<Task<WeatherFlowStartedResponse>>? Start { get; set; }
        public Lazy<Task<WeatherForecastResponse>>? Processing { get; set; }
        public WeatherFlowNotification? Completion { get; set; }
        public bool Delivered { get; set; }
        public bool Sending { get; set; }
        public int DeliveryAttempts { get; set; }
    }
}

namespace dapr_bindings_aspire.ApiService.WeatherForecasting;
 
public interface IWeatherFlowCoordinator
{
    WeatherFlowOptions GetFlowOptions();
 
    Task<WeatherForecastResponse> StoreInputBindingAsync(
        string inputBinding,
        string requestBody,
        CancellationToken cancellationToken);
 
    Task VerifyOutputBindingAsync(
        string outputBinding,
        string requestBody,
        CancellationToken cancellationToken);
 
    Task<WeatherFlowStartedResponse> StartFlowAsync(
        WeatherFlowRequest request,
        CancellationToken cancellationToken);
 
}

using System.Text.Json;
 
namespace dapr_bindings_aspire.ApiService.WeatherForecasting;
 
public sealed class WeatherFlowCoordinator(
    DaprBindingClient dapr,
    InMemoryWeatherFlowStore flows,
    WeatherFlowNotificationDispatcher callbacks,
    ILogger<WeatherFlowCoordinator> logger) : IWeatherFlowCoordinator
{
    private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web);
 
    private static readonly WeatherBindingDefinition[] InputBindings =
    [
        new("rabbitmq", "RabbitMQ", "input-binding-rabbitmq", "trigger-binding-rabbitmq"),
        new("kafka", "Kafka", "input-binding-kafka", "trigger-binding-kafka")
    ];
 
    private static readonly WeatherBindingDefinition[] OutputBindings =
    [
        new("rabbitmq", "RabbitMQ", "output-binding-rabbitmq", null),
        new("kafka", "Kafka", "output-binding-kafka", null),
        new("redis", "Redis", "output-binding-redis", null)
    ];
 
    private static readonly string[] Summaries =
    [
        "Freezing",
        "Bracing",
        "Chilly",
        "Cool",
        "Mild",
        "Warm",
        "Balmy",
        "Hot",
        "Sweltering",
        "Scorching"
    ];
 
 
    public WeatherFlowOptions GetFlowOptions() =>
        new(
            InputBindings.Select(ToOption).ToArray(),
            OutputBindings.Select(ToOption).ToArray());
 
    public Task<WeatherForecastResponse> StoreInputBindingAsync(
        string inputBinding,
        string requestBody,
        CancellationToken cancellationToken)
    {
        var input = FindInput(inputBinding);
        var request = ReadTriggerRequest(requestBody);
        var output = FindOutput(request?.OutputBinding ?? "redis");
        callbacks.ValidateCallback(request?.CallbackUrl);
        var correlationId = string.IsNullOrWhiteSpace(request?.CorrelationId) ? Guid.NewGuid().ToString("N") : request.CorrelationId;
        request = request is null
            ? new(null, null, correlationId, output.Id, null)
            : request with { CorrelationId = correlationId };
        var entry = flows.Register(correlationId, input.Id, output.Id);
        Lazy<Task<WeatherForecastResponse>> processing;
        lock (entry.Gate)
        {
            processing = entry.Processing ??= new(() => ProcessInputAsync(request, input, output, cancellationToken));
        }
        return processing.Value;
    }
 
    private async Task<WeatherForecastResponse> ProcessInputAsync(
        WeatherForecastTriggerRequest request,
        WeatherBindingDefinition input,
        WeatherBindingDefinition output,
        CancellationToken cancellationToken)
    {
        var response = GenerateForecast(request, input, output);
 
        try
        {
            await callbacks.ProgressAsync(
                response,
                FlowStatus.InputReceived,
                $"Dapr delivered the {input.Label} input event to ApiService.",
                cancellationToken);
            await callbacks.ProgressAsync(
                response,
                FlowStatus.ForecastGenerated,
                $"Weather forecast generated; sending it to {output.Label}.",
                cancellationToken);
            await InvokeSelectedOutputAsync(response, output, cancellationToken);
            await callbacks.ProgressAsync(
                response,
                FlowStatus.OutputAccepted,
                $"{output.Label} accepted the weather forecast.",
                cancellationToken);
 
            if (output.Id == "redis")
            {
                response = await VerifyRedisOutputAsync(response, cancellationToken);
                await callbacks.CompleteAsync(response, FlowStatus.Verified, cancellationToken);
            }
            else
            {
                response = response with
                {
                    OutputStatus = $"Weather forecast accepted by {output.Label}; waiting for destination verification."
                };
            }
        }
        catch (Exception ex) when (ex is HttpRequestException or OperationCanceledException or InvalidOperationException
            or Polly.Timeout.TimeoutRejectedException or Polly.CircuitBreaker.BrokenCircuitException)
        {
            logger.LogWarning(
                ex,
                "Weather output binding {BindingName} failed for correlation {CorrelationId}.",
                output.BindingName,
                response.CorrelationId);
 
            response = response with
            {
                OutputStatus = $"{output.Label} output failed: {ex.Message}"
            };
            await callbacks.CompleteAsync(response, FlowStatus.Failed, cancellationToken);
        }
 
        return response;
    }
 
    public async Task VerifyOutputBindingAsync(
        string outputBinding,
        string requestBody,
        CancellationToken cancellationToken)
    {
        var output = FindOutput(outputBinding);
        var response = ReadForecastResponse(requestBody);
        if (response?.CorrelationId is null
            || !string.Equals(response.OutputBinding, output.Id, StringComparison.OrdinalIgnoreCase))
        {
            logger.LogWarning("Ignored unverifiable {BindingName} output message.", output.BindingName);
            return;
        }
 
        if (flows.Find(response.CorrelationId) is not { } entry
            || entry.Output != output.Id || response.InputBinding != entry.Input
            || response.Forecasts is null)
            return;
 
        response = response with
        {
            OutputStatus = $"Weather forecast verified in {output.Label}."
        };
        await callbacks.CompleteAsync(response, FlowStatus.Verified, cancellationToken);
    }
 
    public Task<WeatherFlowStartedResponse> StartFlowAsync(
        WeatherFlowRequest request,
        CancellationToken cancellationToken)
    {
        var input = FindInput(request.InputBinding);
        var output = FindOutput(request.OutputBinding);
        var correlationId = string.IsNullOrWhiteSpace(request.CorrelationId)
            ? Guid.NewGuid().ToString("N")
            : request.CorrelationId;
        callbacks.ValidateCallback(request.CallbackUrl);
        var entry = flows.Register(correlationId, input.Id, output.Id);
        Lazy<Task<WeatherFlowStartedResponse>> start;
        lock (entry.Gate)
        {
            start = entry.Start ??= new(() => PublishTriggerAsync(request, input, output, correlationId, cancellationToken));
        }
        return start.Value;
    }
 
    private async Task<WeatherFlowStartedResponse> PublishTriggerAsync(
        WeatherFlowRequest request, WeatherBindingDefinition input, WeatherBindingDefinition output,
        string correlationId, CancellationToken cancellationToken)
    {
        var trigger = new WeatherForecastTriggerRequest(
            Days: 5,
            StartDate: DateOnly.FromDateTime(DateTime.UtcNow).AddDays(1),
            CorrelationId: correlationId,
            OutputBinding: output.Id,
            CallbackUrl: request.CallbackUrl);
 
        if (input.TriggerBindingName is null)
        {
            throw new InvalidOperationException($"{input.Label} does not have a trigger binding configured.");
        }
 
        {
            using var response = await dapr.InvokeAsync(
                input.TriggerBindingName,
                new DaprBindingRequest(
                    Operation: "create",
                    Data: JsonSerializer.Serialize(trigger, JsonOptions),
                    Metadata: []),
                cancellationToken);
 
            if (!response.IsSuccessStatusCode)
            {
                var details = await response.Content.ReadAsStringAsync(cancellationToken);
                throw new HttpRequestException(
                    $"{input.Label} trigger returned HTTP {(int)response.StatusCode}. {details}");
            }
        }
 
        return new WeatherFlowStartedResponse(
            true,
            $"Flow started through {input.Label}; waiting for Dapr input binding delivery.",
            correlationId,
            input.Id,
            input.Label,
            output.Id,
            output.Label);
    }
 
    private async Task InvokeSelectedOutputAsync(
        WeatherForecastResponse forecastResponse,
        WeatherBindingDefinition output,
        CancellationToken cancellationToken)
    {
        var metadata = output.Id == "redis"
            ? new Dictionary<string, string> { ["key"] = GetRedisKey(forecastResponse.CorrelationId) }
            : [];
 
        using var response = await dapr.InvokeAsync(
            output.BindingName,
            new DaprBindingRequest(
                Operation: "create",
                Data: JsonSerializer.Serialize(forecastResponse, JsonOptions),
                Metadata: metadata),
            cancellationToken);
 
        if (!response.IsSuccessStatusCode)
        {
            var details = await response.Content.ReadAsStringAsync(cancellationToken);
            throw new InvalidOperationException(
                $"{output.Label} returned HTTP {(int)response.StatusCode}. {details}");
        }
 
    }
 
    private async Task<WeatherForecastResponse> VerifyRedisOutputAsync(
        WeatherForecastResponse expected,
        CancellationToken cancellationToken)
    {
        using var response = await dapr.InvokeAsync(
            "output-binding-redis",
            new DaprBindingRequest(
                Operation: "get",
                Data: null,
                Metadata: new Dictionary<string, string> { ["key"] = GetRedisKey(expected.CorrelationId) }),
            cancellationToken);
 
        if (!response.IsSuccessStatusCode)
        {
            var details = await response.Content.ReadAsStringAsync(cancellationToken);
            throw new InvalidOperationException(
                $"Redis verification failed with HTTP {(int)response.StatusCode}. {details}");
        }
 
        var stored = ReadForecastResponse(await response.Content.ReadAsStringAsync(cancellationToken));
        if (stored?.Forecasts is null || stored.OutputBinding != "redis"
            || !string.Equals(stored.CorrelationId, expected.CorrelationId, StringComparison.Ordinal))
        {
            throw new InvalidOperationException("Redis verification did not return the forecast written for this flow.");
        }
 
        return stored with
        {
            OutputStatus = "Weather forecast verified in Redis."
        };
    }
 
    private WeatherForecastResponse GenerateForecast(
        WeatherForecastTriggerRequest? request,
        WeatherBindingDefinition input,
        WeatherBindingDefinition output)
    {
        var days = Math.Clamp(request?.Days ?? 5, 1, 10);
        var startDate = request?.StartDate ?? DateOnly.FromDateTime(DateTime.UtcNow).AddDays(1);
        var forecasts = Enumerable.Range(0, days)
            .Select(index => new WeatherForecast(
                startDate.AddDays(index),
                Random.Shared.Next(-20, 55),
                Summaries[Random.Shared.Next(Summaries.Length)]))
            .ToArray();
 
        return new WeatherForecastResponse(
            DateTimeOffset.UtcNow,
            forecasts,
            string.IsNullOrWhiteSpace(request?.CorrelationId) ? Guid.NewGuid().ToString("N") : request.CorrelationId,
            input.Id,
            input.Label,
            output.Id,
            output.Label);
    }
 
    private static WeatherForecastTriggerRequest? ReadTriggerRequest(string requestBody)
    {
        if (string.IsNullOrWhiteSpace(requestBody))
        {
            return null;
        }
 
        try
        {
            using var document = JsonDocument.Parse(requestBody);
            var root = document.RootElement;
 
            if (root.ValueKind == JsonValueKind.Object && root.TryGetProperty("data", out var data))
            {
                return ReadTriggerRequest(data);
            }
 
            return ReadTriggerRequest(root);
        }
        catch (JsonException)
        {
            return null;
        }
    }
 
    private static WeatherForecastTriggerRequest? ReadTriggerRequest(JsonElement element)
    {
        if (element.ValueKind == JsonValueKind.String)
        {
            return ReadTriggerRequest(element.GetString() ?? string.Empty);
        }
 
        return element.Deserialize<WeatherForecastTriggerRequest>(JsonOptions);
    }
 
    private static WeatherForecastResponse? ReadForecastResponse(string content)
    {
        if (string.IsNullOrWhiteSpace(content))
        {
            return null;
        }
 
        try
        {
            using var document = JsonDocument.Parse(content);
            return ReadForecastResponse(document.RootElement);
        }
        catch (JsonException)
        {
            return null;
        }
    }
 
    private static WeatherForecastResponse? ReadForecastResponse(JsonElement element)
    {
        if (element.ValueKind == JsonValueKind.Object && element.TryGetProperty("data", out var data))
        {
            return ReadForecastResponse(data);
        }
 
        if (element.ValueKind == JsonValueKind.String)
        {
            var value = element.GetString();
            if (string.IsNullOrWhiteSpace(value))
            {
                return null;
            }
 
            return ReadForecastResponse(value);
        }
 
        return element.Deserialize<WeatherForecastResponse>(JsonOptions);
    }
 
    private static string GetRedisKey(string? correlationId) =>
        $"weatherforecast:{correlationId ?? throw new InvalidOperationException("A correlation ID is required for Redis verification.")}";
 
    private static WeatherBindingOption ToOption(WeatherBindingDefinition definition) =>
        new(definition.Id, definition.Label, definition.BindingName);
 
    private static WeatherBindingDefinition FindInput(string id) =>
        InputBindings.FirstOrDefault(binding => string.Equals(binding.Id, id, StringComparison.OrdinalIgnoreCase))
        ?? throw new FlowValidationException($"Unknown input binding '{id}'.");
 
    private static WeatherBindingDefinition FindOutput(string id) =>
        OutputBindings.FirstOrDefault(binding => string.Equals(binding.Id, id, StringComparison.OrdinalIgnoreCase))
        ?? throw new FlowValidationException($"Unknown output binding '{id}'.");
 
    private sealed record WeatherBindingDefinition(
        string Id,
        string Label,
        string BindingName,
        string? TriggerBindingName);
}

using System.Net.Http.Json;
 
namespace dapr_bindings_aspire.ApiService.WeatherForecasting;
 
public sealed class WeatherFlowNotificationDispatcher(
    IHttpClientFactory clients,
    IConfiguration configuration,
    InMemoryWeatherFlowStore registry,
    ILogger<WeatherFlowNotificationDispatcher> logger) : BackgroundService
{
    public const string DefaultUrl = "https+http://webfrontend/weather-flow-notifications";
    public const int MaxAttempts = 10;
    private string CallbackUrl => configuration["WeatherFlow:CallbackUrl"] ?? DefaultUrl;
 
    public void ValidateCallback(string? requestedUrl)
    {
        if (!string.IsNullOrWhiteSpace(requestedUrl) && !string.Equals(requestedUrl, CallbackUrl, StringComparison.Ordinal))
            throw new FlowValidationException("Callback URL must match the configured Web notification endpoint.");
    }
 
    public Task ProgressAsync(WeatherForecastResponse forecast, string status, string message, CancellationToken cancellationToken) =>
        SendAsync(new(forecast.CorrelationId!, status, message, forecast), cancellationToken);
 
    public async Task CompleteAsync(WeatherForecastResponse forecast, string status, CancellationToken cancellationToken)
    {
        if (forecast.CorrelationId is null || registry.Find(forecast.CorrelationId) is not { } entry)
            return;
        lock (entry.Gate)
        {
            entry.Completion ??= new(forecast.CorrelationId, status, forecast.OutputStatus ?? "Flow ended.", forecast);
        }
        logger.LogInformation("Flow {CorrelationId} reached {Status} for {OutputBinding}.",
            forecast.CorrelationId, status, forecast.OutputBinding);
        await DeliverAsync(entry, cancellationToken);
    }
 
    public async Task RetryPendingAsync(CancellationToken cancellationToken)
    {
        await Parallel.ForEachAsync(registry.Snapshot(), new ParallelOptions
        {
            MaxDegreeOfParallelism = 4,
            CancellationToken = cancellationToken
        }, async (entry, token) => await DeliverAsync(entry, token));
    }
 
    private async Task DeliverAsync(InMemoryWeatherFlowStore.Entry entry, CancellationToken cancellationToken)
    {
        WeatherFlowNotification notification;
        lock (entry.Gate)
        {
            if (entry.Completion is null || entry.Delivered || entry.Sending || entry.DeliveryAttempts >= MaxAttempts)
                return;
            notification = entry.Completion;
            entry.Sending = true;
            entry.DeliveryAttempts++;
        }
        try
        {
            var delivered = await SendAsync(notification, cancellationToken);
            lock (entry.Gate)
                entry.Delivered = delivered;
            if (delivered)
                logger.LogInformation("Completion callback delivered for {CorrelationId} with {Status}.",
                    notification.CorrelationId, notification.Status);
            else if (entry.DeliveryAttempts >= MaxAttempts)
                logger.LogWarning("Completion callback exhausted {Attempts} attempts for {CorrelationId}.",
                    MaxAttempts, notification.CorrelationId);
        }
        finally
        {
            lock (entry.Gate)
                entry.Sending = false;
        }
    }
 
    private async Task<bool> SendAsync(WeatherFlowNotification notification, CancellationToken cancellationToken)
    {
        var token = configuration["WeatherFlow:CallbackToken"];
        if (string.IsNullOrWhiteSpace(token))
        {
            logger.LogWarning("Callback token is not configured; notification for {CorrelationId} was not sent.", notification.CorrelationId);
            return false;
        }
        using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        timeout.CancelAfter(TimeSpan.FromSeconds(2));
        try
        {
            using var request = new HttpRequestMessage(HttpMethod.Post, CallbackUrl)
            {
                Content = JsonContent.Create(notification)
            };
            request.Headers.Add("X-Weather-Callback-Token", token);
            using var response = await clients.CreateClient("weather-notifications").SendAsync(request, timeout.Token);
            if (response.IsSuccessStatusCode)
                return true;
            logger.LogWarning("Callback returned {StatusCode} for {CorrelationId}.", (int)response.StatusCode, notification.CorrelationId);
        }
        catch (Exception ex) when (ex is HttpRequestException or OperationCanceledException
            or Polly.Timeout.TimeoutRejectedException or Polly.CircuitBreaker.BrokenCircuitException)
        {
            logger.LogWarning(ex, "Callback delivery failed for {CorrelationId}.", notification.CorrelationId);
        }
        return false;
    }
 
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(2));
        try
        {
            while (await timer.WaitForNextTickAsync(stoppingToken))
                await RetryPendingAsync(stoppingToken);
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { }
    }
}

aspire run