User Defined Wire-Message Mapping

**Is your feature request related to a problem? Please describe.**
We currently write the trace context information into message headers after the message mapping pipeline runs. This doesn't allow anyone to modify where we decide to place it when mapping to an on-the-wire message. This works for CloudEvents-based environments, where we write the MessageHeader parent and baggage properties to the CloudEvents headers.

But this doesn't work for interop with environments that do support telemetry but don't support Cloud Event, which would like to be able to decide where to write this, for example, in our bag or in the body.

Because we choose how to serialize our internal Message class to the wire representation, you can't control that directly.

To write the trace context information into the Message Headers, we use “src/Paramore.Brighter/Observability/TextContextPropogator.cs”. The `TextContextPropogator` is a standard pattern that we implement via `void PropogateContext(ActivityContext? context, Message message);`. The `PropogateContext` method invokes the “src/Paramore.Brighter/Message.cs” method “public static void PropogateContext(Message message, string key, string? value)”. We use this to set the message’s headers.

```
public static void PropogateContext(Message message, string key, string? value)
{
if (value is null)
return;

switch (key)
{
case "traceparent":
message.Header.TraceParent = value;
break;
case "tracestate":
message.Header.TraceState = value;
break;
case "baggage":
message.Header.Baggage.LoadBaggage(value);
break;
}
}


The `TextContextPropogator` is called from `BrighterTracer` (see src/Paramore.Brighter/Observability/BrighterTracer.cs), a class that contains our tracing code in one place, reducing complexity at the call site. It is called when we produce a message via `public static void WriteProducerEvent(Activity? span, string messagingSystem, Message message, InstrumentationOptions instrumentationOptions)`.

`BrighterTracer.WriteProductEvent` is itself called from an implementation of `IAmAMessageProducerSync` or `IAmAMessageProducerAsync` on their `SendWithDelayAsync` method. This, in turn, is called from our `OutboxProducerMediator` (see src/Paramore.Brighter/OutboxProducerMediator.cs), which manages Outbox-based sending to a MessagingGateway to send a message via its `private void Dispatch(IEnumerable posts, RequestContext requestContext, Dictionary? args = null)` method, which is, in turn, called from `public void ClearOutbox`. This `ClearOutbox` method is called from the `CommandProcessor`’s own `public void ClearOutbox(Id[] ids, RequestContext? requestContext = null, Dictionary? args = null)` (see src/Paramore.Brighter/CommandProcessor.cs).

The intent of `ClearOutbox` is to pull messages from the Outbox that have not yet been dispatched and send them either directly or via a background sweeper process.

For this account, it's worth noting that what we pull from the Outbox is the message we intend to send. It is worth understanding that in this context, the message has already been written.

Where then was it written?

The message is written via the `CommandProcessor`’s `DepositPost` method. `DepositPost` works with the `OutboxProducerMediator` to call `public Message CreateMessageFromRequest(TRequest request, RequestContext requestContext)`. This method calls the internal `private Message MapMessage(TRequest request, RequestContext requestContext)`, which creates the message mapping pipeline and any transformer middleware.

As a result, the message transformation and mapping pipeline runs when creating the message and cannot alter the wire format and where we place the trace information, since that is decided after we have created and stored the message on disk.

Let’s assume that in any given implementation of a MessagingGateway, we have code that looks akin to this (src/Paramore.Brighter.MessagingGateway.AWSSQS.V4/SqsMessageSender.cs) to actually create the wire format message (in this case SQS) from a Brighter `Message`:

```
private void SetMessageAttributes(SendMessageRequest request, Message message)
{
string cloudEventHeadersJson = CreateCloudEventHeadersJson(message);

var contentType = message.Header.ContentType ?? new ContentType(MediaTypeNames.Text.Plain);
var messageAttributes = new Dictionary
{
[HeaderNames.Id] = new() { StringValue = message.Header.MessageId, DataType = "String" },
[HeaderNames.CloudEventHeaders] = new() { StringValue = cloudEventHeadersJson, DataType = "String" },
[HeaderNames.Topic] = new() { StringValue = _queueUrl, DataType = "String" },
[HeaderNames.MessageType] = new() { StringValue = message.Header.MessageType.ToString(), DataType = "String" },
[HeaderNames.ContentType] = new() { StringValue = contentType.ToString(), DataType = "String" },
[HeaderNames.Timestamp] = new() { StringValue = Convert.ToString(message.Header.TimeStamp.ToRfc3339()), DataType = "String" }
};

if (!RoutingKey.IsNullOrEmpty(message.Header.ReplyTo))
messageAttributes.Add(HeaderNames.ReplyTo, new MessageAttributeValue { StringValue = message.Header.ReplyTo, DataType = "String" });

if (!string.IsNullOrEmpty(message.Header.Subject))
messageAttributes.Add(HeaderNames.Subject, new MessageAttributeValue { StringValue = message.Header.Subject, DataType = "String" });

if (!Id.IsNullOrEmpty(message.Header.CorrelationId))
messageAttributes.Add(HeaderNames.CorrelationId, new MessageAttributeValue { StringValue = message.Header.CorrelationId, DataType = "String" });

message.Header.Bag[HeaderNames.HandledCount] = message.Header.HandledCount.ToString(CultureInfo.InvariantCulture);

var bagJson = System.Text.Json.JsonSerializer.Serialize(message.Header.BagWithoutLocalHeaders(), JsonSerialisationOptions.Options);
messageAttributes[HeaderNames.Bag] = new() { StringValue = bagJson, DataType = "String" };
request.MessageAttributes = messageAttributes;
}

What we have are our hardcoded assumptions about how Brighter’s own `MessageHeader` maps to metadata on the wire (for now, we assume the body-mapping issues are safely handled by the Transform -> MessageMapper pipeline). We assume that we want to use a binary Cloud Events mapping, and if not a structured Cloud Events mapping, alongside some useful Brighter proprties (that are not required). This is the weak part of our design. There is no way to override how metadata is mapped if you want to support something else. We allow alteration of how the mapping to and from a request works, but not alteration of how mapping to and from the wire works.

Now we need to talk about the reverse, mapping back from an on-the-wire message. But this again has the same limitation. If we receive a message on the wire, our code maps it to the internal `Message` type. The structure of the code here is “by convention” rather than defined roles (interfaces or abstract base types) to implement (this may change in the future as we look to a factory approach, which is more amenable to agentic engineering).

As an example, though, let’s follow SQS mapping, which has triggered this work.

`SqsMessageCreator` is the class that reads the message's wire representation and converts it into a `Message`. It uses our option-type convention, HeaderResult, to do this.

```
topic = ReadTopic(sqsMessage);
messageId = ReadMessageId(sqsMessage);
var cloudEventHeaders = ReadCloudEventHeaders(sqsMessage);
var bag = ReadMessageBag(sqsMessage);
var contentType = ReadContentType(sqsMessage, cloudEventHeaders);
var correlationId = ReadCorrelationId(sqsMessage);
var messageType = ReadMessageType(sqsMessage);
var timeStamp = ReadTimestamp(sqsMessage, cloudEventHeaders);
var replyTo = ReadReplyTo(sqsMessage);
var receiptHandle = ReadReceiptHandle(sqsMessage);
var partitionKey = ReadPartitionKey(sqsMessage);
var deduplicationId = ReadDeduplicationId(sqsMessage);
var subject = ReadSubject(sqsMessage, cloudEventHeaders);
var handledCount = ReadHandledCount(bag);
var source = ReadCloudEventSource(cloudEventHeaders);
var type = ReadCloudEventType(cloudEventHeaders);
var dataSchema = ReadCloudEventsDataSchema(cloudEventHeaders);
var specVersion = ReadCloudEventsSpecVersion(cloudEventHeaders);
var traceParent = ReadCloudEventsTraceParent(cloudEventHeaders);
var traceState = ReadCloudEventsTraceState(cloudEventHeaders);
var baggage = ReadCloudEventsBaggage(cloudEventHeaders);

var bodyType = contentType.Success ? contentType.Result : new ContentType(MediaTypeNames.Text.Plain);

```

HeaderResult means that we provide sensible defaults in the event of missing values, but the methods called assume the location where we read from:

```
private static HeaderResult ReadCloudEventsTraceParent(Dictionary cloudEventHeaders)
{
if (cloudEventHeaders.TryGetValue(HeaderNames.TraceParent, out var value))
{
return new HeaderResult(new TraceParent(value), true);
}

return new HeaderResult(null, true);
}
```

What creates a problem is when they are not at the location but are in the metadata of the incoming message. We just won’t read them, but will present a default or empty `MessageHeader` property in their place.

How then do we allow for different mappings to a wire format, other than Cloud Events?

**Who will use this?**


We expect two audiences for this. The first is internal, where we choose to provide integration with a third party and need to map; the second is external, where an outside company depends on integrations (such as non-.NET languages) that we don’t support and needs to provide a way to map attributes from the wire.

**Describe the solution you'd like**

Two options look possible here, both are variations of the same idea as the MessageMapper pipeline - allow the mapping, but this time of wire format to Brighter `Message` (as opposed to Brighter `Message` to request) to come under user control if they don’t want our default.

(1) Turn the existing mappings into a default implementation of a Func<,> which provides for how to map, and allow this to be overridden by a Func<,> registered at startup, presumably via a property on the publication/subscription, if someone wants to change our mappings
(2) Provide a DSL for the mapping, and allow this to be passed in, again via a property on the publication/subscription, to override the default existing mapping.

Overall, 1 is a greater burden for the end user to get right, and 2 is a greater burden for the Brighter team to get right. We would favor the latter, given that configuration is already a burden for choosing something like Brighter over bespoke code. For most users, this ought to be invisible, an advanced feature they never need. But for us, and for some partners, it lets integration work

We may be able to use an [internal DSL](https://martinfowler.com/bliki/InternalDslStyle.html) in C# here, rather than an external one, to describe the mapping, which would make this easier. Conceptually, we imagine something like [bloblang](https://docs.redpanda.com/connect/guides/bloblang/about/), a simple mapping language, or something using [Object Scoping](https://martinfowler.com/dslCatalog/objectScoping.html) (a base class that exposes a range of methods available as a DSL in a derived class).

There may be alternative designs that let us achieve this goal, allowing variation in how we map from a Brighter message to the wire representation.

**Describe alternatives you've considered**
The message mapper pipeline has sufficient information to propogate the trace context, flowed via the RequestContext and it would be possible to write a Transform (see “src/Paramore.Brighter/IAmAMessageTransform.cs” or “src/Paramore.Brighter/IAmAMessageTransformAsync.cs”) for a specific legacy use case that sets trace context in the bag, such that it could be flowed from there. The advantage of this approach is that we continue to have a” pit of success” around trace propagation for CloudEvents scenarios, but also have a backup ability to write a Transform when a legacy scenario requires it in the bag.e 

The limitation here is that the bag may be serialized in one (as JSON), and so if the legacy client created it under a specific value, we don’t directly support that. In that case, we don’t have a direct means to remap to that transport that is under control.

So we are no closer to a solution from a Transform because we don’t have a mapping at the MessagingGateway level for a specific provider.

Really, then, the binding's lateness matters less than the fact that we have no control over how we bind `Message` to the underlying protocol.

Another option would be to write a MessagingGateway specifically for that legacy scenario, but this seems a significant effort, just to support a system that defined its headers before CloudEvents.

**Additional context**
Add any other context or screenshots about the feature request here.