Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 64 additions & 0 deletions src/Dapr.Client/DaprOutboxMetadata.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
// ------------------------------------------------------------------------
// Copyright 2026 The Dapr Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
// http://www.apache.org/licenses/LICENSE-2.0
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------------

#nullable enable

namespace Dapr.Client;

/// <summary>
/// Well-known metadata keys used by Dapr's transactional outbox feature on
/// state transaction operations and by the <see cref="OutboxTransactionBuilder"/>.
/// </summary>
Comment thread
WhitWaldo marked this conversation as resolved.
/// <remarks>
/// See <a href="https://docs.dapr.io/developing-applications/building-blocks/state-management/howto-outbox/">
/// How-To: Enable the transactional outbox pattern</a> for details.
/// </remarks>
public static class DaprOutboxMetadata
{
/// <summary>
/// Marks a <see cref="StateTransactionRequest"/> as an outbox projection.
/// The item is not written to the state store; its value is used as the payload
/// published to the pub/sub topic configured on the state store component.
/// </summary>
public const string Projection = "outbox.projection";

/// <summary>
/// Value used with <see cref="Projection"/> to enable the projection behavior.
/// </summary>
public const string ProjectionEnabled = "true";

/// <summary>
/// Overrides the CloudEvent <c>id</c> field on the published outbox event.
/// </summary>
public const string CloudEventId = "cloudevent.id";

/// <summary>
/// Overrides the CloudEvent <c>source</c> field on the published outbox event.
/// </summary>
public const string CloudEventSource = "cloudevent.source";

/// <summary>
/// Overrides the CloudEvent <c>type</c> field on the published outbox event.
/// </summary>
public const string CloudEventType = "cloudevent.type";

/// <summary>
/// Overrides the CloudEvent <c>subject</c> field on the published outbox event.
/// </summary>
public const string CloudEventSubject = "cloudevent.subject";

/// <summary>
/// Overrides the CloudEvent <c>datacontenttype</c> field on the published outbox event.
/// </summary>
public const string CloudEventDataContentType = "cloudevent.datacontenttype";
}
153 changes: 153 additions & 0 deletions src/Dapr.Client/OutboxTransactionBuilder.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,153 @@
// ------------------------------------------------------------------------
// Copyright 2026 The Dapr Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
// http://www.apache.org/licenses/LICENSE-2.0
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------------

#nullable enable
using System;
using System.Collections.Generic;

namespace Dapr.Client;

/// <summary>
/// Fluent builder that produces a list of <see cref="StateTransactionRequest"/> items
/// shaped for Dapr's transactional outbox feature. Each entry pairs the state operation
/// that is persisted to the state store with an optional projection that shapes the
/// payload published to the pub/sub topic.
/// </summary>
/// <remarks>
/// Dapr's outbox requires the projection key to match the state operation key exactly.
/// This builder enforces that invariant at build time.
/// </remarks>
public sealed class OutboxTransactionBuilder
{
private readonly List<Entry> entries = new();

/// <summary>
/// Adds a state upsert whose value is also published as the outbox event.
/// </summary>
/// <param name="key">The state key.</param>
/// <param name="value">The serialized state value, written to the store and used as the published payload.</param>
/// <param name="etag">Optional ETag for optimistic concurrency.</param>
/// <param name="metadata">Optional metadata applied to the state operation.</param>
/// <param name="options">Optional state options.</param>
/// <returns>This builder, for chaining.</returns>
public OutboxTransactionBuilder Upsert(
string key,
byte[] value,
string? etag = null,
IReadOnlyDictionary<string, string>? metadata = null,
StateOptions? options = null)
{
ArgumentException.ThrowIfNullOrEmpty(key);
ArgumentNullException.ThrowIfNull(value);

entries.Add(new Entry(
new StateTransactionRequest(key, value, StateOperationType.Upsert, etag, metadata, options),
Projection: null));

return this;
}

/// <summary>
/// Adds a state upsert paired with an explicit projection that shapes the outbox event
/// payload independently of the state value.
/// </summary>
/// <param name="key">The state key. The projection uses the same key.</param>
/// <param name="stateValue">The value persisted to the state store.</param>
/// <param name="projectionValue">The value published as the outbox event payload.</param>
/// <param name="etag">Optional ETag for optimistic concurrency on the state write.</param>
/// <param name="stateMetadata">Optional metadata applied to the state operation.</param>
/// <param name="projectionMetadata">
/// Optional metadata applied to the projection (for example, CloudEvent overrides from
/// <see cref="DaprOutboxMetadata"/>). The <see cref="DaprOutboxMetadata.Projection"/> key
/// is always added.
/// </param>
/// <param name="options">Optional state options for the state write.</param>
/// <returns>This builder, for chaining.</returns>
public OutboxTransactionBuilder UpsertWithProjection(
string key,
byte[] stateValue,
byte[] projectionValue,
string? etag = null,
IReadOnlyDictionary<string, string>? stateMetadata = null,
IReadOnlyDictionary<string, string>? projectionMetadata = null,
StateOptions? options = null)
{
ArgumentException.ThrowIfNullOrEmpty(key);
ArgumentNullException.ThrowIfNull(stateValue);
ArgumentNullException.ThrowIfNull(projectionValue);

var state = new StateTransactionRequest(key, stateValue, StateOperationType.Upsert, etag, stateMetadata, options);
var projection = new StateTransactionRequest(key, projectionValue, StateOperationType.Upsert, etag: null, metadata: projectionMetadata, options: null)
.WithOutboxProjection();

entries.Add(new Entry(state, projection));

return this;
}

/// <summary>
/// Adds a state delete whose absence is signalled to the outbox topic.
/// </summary>
/// <param name="key">The state key.</param>
/// <param name="etag">Optional ETag for optimistic concurrency.</param>
/// <param name="metadata">Optional metadata applied to the state operation.</param>
/// <param name="options">Optional state options.</param>
/// <returns>This builder, for chaining.</returns>
public OutboxTransactionBuilder Delete(
string key,
string? etag = null,
IReadOnlyDictionary<string, string>? metadata = null,
StateOptions? options = null)
{
ArgumentException.ThrowIfNullOrEmpty(key);

entries.Add(new Entry(
new StateTransactionRequest(key, value: null, StateOperationType.Delete, etag, metadata, options),
Projection: null));

return this;
}

/// <summary>
/// Builds the flat list of <see cref="StateTransactionRequest"/> items in the order
/// suitable for passing to <see cref="DaprClient.ExecuteStateTransactionAsync"/>.
/// State operations always precede their paired projections.
/// </summary>
/// <returns>An immutable list of transaction requests.</returns>
public IReadOnlyList<StateTransactionRequest> Build()
{
var result = new List<StateTransactionRequest>(entries.Count * 2);
for (var i = 0; i < entries.Count; i++)
{
var entry = entries[i];
result.Add(entry.State);
if (entry.Projection is not null)
{
if (!string.Equals(entry.State.Key, entry.Projection.Key, StringComparison.Ordinal))
{
// Defensive: internally we always construct projections with the same key,
// so this should never happen. Guard prevents silent corruption if the
// internal invariant ever breaks.
throw new InvalidOperationException(
$"Outbox projection key '{entry.Projection.Key}' does not match state key '{entry.State.Key}'.");
}

result.Add(entry.Projection);
}
}

return result;
}

private readonly record struct Entry(StateTransactionRequest State, StateTransactionRequest? Projection);
}
141 changes: 141 additions & 0 deletions src/Dapr.Client/StateTransactionRequestExtensions.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
// ------------------------------------------------------------------------
// Copyright 2026 The Dapr Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
// http://www.apache.org/licenses/LICENSE-2.0
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------------

#nullable enable
using System;
using System.Collections.Generic;

namespace Dapr.Client;

/// <summary>
/// Extension methods for shaping <see cref="StateTransactionRequest"/> instances
/// for use with Dapr's transactional outbox feature.
/// </summary>
public static class StateTransactionRequestExtensions
{
/// <summary>
/// Returns a new <see cref="StateTransactionRequest"/> whose metadata marks it as an
/// outbox projection (see <see cref="DaprOutboxMetadata.Projection"/>). The source
/// request is not mutated.
/// </summary>
/// <param name="request">The state transaction request to convert into a projection.</param>
/// <returns>A new <see cref="StateTransactionRequest"/> with the projection metadata set.</returns>
public static StateTransactionRequest WithOutboxProjection(this StateTransactionRequest request)
{
ArgumentNullException.ThrowIfNull(request);

var merged = MergeMetadata(request.Metadata, new KeyValuePair<string, string>[]
{
new(DaprOutboxMetadata.Projection, DaprOutboxMetadata.ProjectionEnabled),
});

return new StateTransactionRequest(
request.Key,
request.Value,
request.OperationType ?? StateOperationType.Upsert,
request.ETag,
merged,
request.Options);
}

/// <summary>
/// Returns a new <see cref="StateTransactionRequest"/> with the specified CloudEvent
/// fields set via metadata. Existing metadata is preserved; only the supplied fields
/// are overridden. The source request is not mutated.
/// </summary>
/// <param name="request">The state transaction request to enrich.</param>
/// <param name="id">The CloudEvent <c>id</c> field, or <see langword="null"/> to leave unchanged.</param>
/// <param name="source">The CloudEvent <c>source</c> field, or <see langword="null"/> to leave unchanged.</param>
/// <param name="type">The CloudEvent <c>type</c> field, or <see langword="null"/> to leave unchanged.</param>
/// <param name="subject">The CloudEvent <c>subject</c> field, or <see langword="null"/> to leave unchanged.</param>
/// <param name="dataContentType">The CloudEvent <c>datacontenttype</c> field, or <see langword="null"/> to leave unchanged.</param>
/// <returns>A new <see cref="StateTransactionRequest"/> with the CloudEvent metadata merged in.</returns>
public static StateTransactionRequest WithCloudEventOverrides(
this StateTransactionRequest request,
string? id = null,
string? source = null,
string? type = null,
string? subject = null,
string? dataContentType = null)
{
ArgumentNullException.ThrowIfNull(request);

var overrides = new List<KeyValuePair<string, string>>(capacity: 5);
if (id is not null)
{
overrides.Add(new KeyValuePair<string, string>(DaprOutboxMetadata.CloudEventId, id));
}
if (source is not null)
{
overrides.Add(new KeyValuePair<string, string>(DaprOutboxMetadata.CloudEventSource, source));
}
if (type is not null)
{
overrides.Add(new KeyValuePair<string, string>(DaprOutboxMetadata.CloudEventType, type));
}
if (subject is not null)
{
overrides.Add(new KeyValuePair<string, string>(DaprOutboxMetadata.CloudEventSubject, subject));
}
if (dataContentType is not null)
{
overrides.Add(new KeyValuePair<string, string>(DaprOutboxMetadata.CloudEventDataContentType, dataContentType));
}

if (overrides.Count == 0)
{
// Nothing to change; still return a new instance to keep the API purely non-mutating
// regardless of caller expectations.
return new StateTransactionRequest(
request.Key,
request.Value,
request.OperationType ?? StateOperationType.Upsert,
request.ETag,
request.Metadata,
request.Options);
}

var merged = MergeMetadata(request.Metadata, overrides);

return new StateTransactionRequest(
request.Key,
request.Value,
request.OperationType ?? StateOperationType.Upsert,
request.ETag,
merged,
request.Options);
}

private static IReadOnlyDictionary<string, string> MergeMetadata(
IReadOnlyDictionary<string, string>? existing,
IReadOnlyList<KeyValuePair<string, string>> overrides)
{
var merged = new Dictionary<string, string>(StringComparer.Ordinal);

if (existing is not null)
{
foreach (var kvp in existing)
{
merged[kvp.Key] = kvp.Value;
}
}

for (var i = 0; i < overrides.Count; i++)
{
var kvp = overrides[i];
merged[kvp.Key] = kvp.Value;
}

return merged;
}
}
35 changes: 35 additions & 0 deletions test/Dapr.Client.Test/DaprOutboxMetadataTests.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
// ------------------------------------------------------------------------
// Copyright 2026 The Dapr Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
// http://www.apache.org/licenses/LICENSE-2.0
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------------

namespace Dapr.Client.Test;

using Shouldly;
using Xunit;

public class DaprOutboxMetadataTests
{
[Fact]
public void Constants_MatchDaprRuntimeContract()
{
// These string values are the wire contract used by the Dapr sidecar's
// outbox implementation. If they change, the SDK stops interoperating
// with the runtime. This test locks them.
DaprOutboxMetadata.Projection.ShouldBe("outbox.projection");
DaprOutboxMetadata.ProjectionEnabled.ShouldBe("true");
DaprOutboxMetadata.CloudEventId.ShouldBe("cloudevent.id");
DaprOutboxMetadata.CloudEventSource.ShouldBe("cloudevent.source");
DaprOutboxMetadata.CloudEventType.ShouldBe("cloudevent.type");
DaprOutboxMetadata.CloudEventSubject.ShouldBe("cloudevent.subject");
DaprOutboxMetadata.CloudEventDataContentType.ShouldBe("cloudevent.datacontenttype");
}
}
Loading
Loading