Skip to content

Rest of transfer flow now uses outbox #680

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tnickelsen wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Rest of transfer flow now uses outbox #680

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e55b2df
Rest of transfer flow now uses outbox
tnickelsen Aug 25, 2025
82fe601
fixed build
tnickelsen Aug 25, 2025
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
106 changes: 99 additions & 7 deletions src/ProjectOrigin.Vault/EventHandlers/VaultSendRegistryTransactionConsumer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,21 +1,26 @@
using Google.Protobuf;
using Grpc.Net.Client;
using MassTransit;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ProjectOrigin.Electricity.V1;
using ProjectOrigin.HierarchicalDeterministicKeys.Interfaces;
using ProjectOrigin.PedersenCommitment;
using ProjectOrigin.Registry.V1;
using ProjectOrigin.Vault.Database;
using ProjectOrigin.Vault.Extensions;
using ProjectOrigin.Vault.Models;
using ProjectOrigin.Vault.Options;
using System;
using System.Linq;
using System.Security.Cryptography;
using System.Text.Json;
using System.Threading.Tasks;
using ProjectOrigin.Vault.Database;

namespace ProjectOrigin.Vault.EventHandlers;

public record TransferFullSliceRegistryTransactionArguments
{
public required Transaction Transaction { get; init; }
public required string RegistryName { get; set; }
public required Guid CertificateId { get; set; }
public required Guid SliceId { get; set; }
Expand All @@ -27,14 +32,14 @@

public record TransferPartialSliceRegistryTransactionArguments
{
public required Transaction Transaction { get; init; }
public required string RegistryName { get; set; }
public required Guid CertificateId { get; set; }
public required Guid TransferredSliceId { get; set; }
public required Guid RemainderSliceId { get; set; }
public required Guid SourceSliceId { get; set; }
public required WalletAttribute[] WalletAttributes { get; set; }
public required Guid ExternalEndpointId { get; set; }
public required uint Quantity { get; set; }
public RequestStatusArgs? RequestStatusArgs { get; set; }
}

Expand All @@ -60,15 +65,24 @@

try
{
await SendTransactionToRegistry(msg.Transaction);
var externalEndpoint = await _unitOfWork.WalletRepository.GetExternalEndpoint(msg.ExternalEndpointId);
var sourceSlice = await _unitOfWork.CertificateRepository.GetWalletSlice(msg.SliceId);
var nextReceiverPosition = await _unitOfWork.WalletRepository.GetNextNumberForId(externalEndpoint.Id);
var receiverPublicKey = externalEndpoint.PublicKey.Derive(nextReceiverPosition).GetPublicKey();

var transferredEvent = CreateTransferEvent(sourceSlice, receiverPublicKey);
var sourceSlicePrivateKey = await _unitOfWork.WalletRepository.GetPrivateKeyForSlice(sourceSlice.Id);
var transaction = sourceSlicePrivateKey.SignRegistryTransaction(transferredEvent.CertificateId, transferredEvent);

await SendTransactionToRegistry(transaction);

var full = new TransferFullSliceWaitCommittedTransactionArguments
{
CertificateId = msg.CertificateId,
RegistryName = msg.RegistryName,
SliceId = msg.SliceId,
TransferredSliceId = msg.TransferredSliceId,
TransactionId = msg.Transaction.ToShaId(),
TransactionId = transaction.ToShaId(),
ExternalEndpointId = msg.ExternalEndpointId,
RequestStatusArgs = msg.RequestStatusArgs,
WalletAttributes = msg.WalletAttributes
Expand All @@ -84,7 +98,7 @@

_logger.LogInformation("Ending consumer: {Consumer} with arguments {Args}", nameof(VaultSendRegistryTransactionConsumer), nameof(TransferFullSliceRegistryTransactionArguments));
}
catch (Exception ex)

Check warning on line 101 in src/ProjectOrigin.Vault/EventHandlers/VaultSendRegistryTransactionConsumer.cs

View workflow job for this annotation

GitHub Actions / analyse / sonar-analysis 

Either log this exception and handle it, or rethrow it with some contextual information. (https://rules.sonarsource.com/csharp/RSPEC-2139)

Check warning on line 101 in src/ProjectOrigin.Vault/EventHandlers/VaultSendRegistryTransactionConsumer.cs

View workflow job for this annotation

GitHub Actions / analyse / sonar-analysis 

Either log this exception and handle it, or rethrow it with some contextual information. (https://rules.sonarsource.com/csharp/RSPEC-2139)
{
_logger.LogError(ex, "Error sending transactions to registry");
throw;
Expand All @@ -99,7 +113,27 @@

try
{
await SendTransactionToRegistry(msg.Transaction);
var sourceSlice = await _unitOfWork.CertificateRepository.GetWalletSlice(msg.SourceSliceId);

var quantity = msg.Quantity;
var remainder = (uint)sourceSlice.Quantity - quantity;

var receiverEndpoints = await _unitOfWork.WalletRepository.GetExternalEndpoint(msg.ExternalEndpointId);
var receiverPosition = await _unitOfWork.WalletRepository.GetNextNumberForId(receiverEndpoints.Id);
var receiverPublicKey = receiverEndpoints.PublicKey.Derive(receiverPosition).GetPublicKey();
var receiverCommitment = new SecretCommitmentInfo(quantity);

var sourceEndpoint = await _unitOfWork.WalletRepository.GetWalletEndpoint(sourceSlice.WalletEndpointId);
var remainderEndpoint = await _unitOfWork.WalletRepository.GetWalletRemainderEndpoint(sourceEndpoint.WalletId);
var remainderPosition = await _unitOfWork.WalletRepository.GetNextNumberForId(remainderEndpoint.Id);
var remainderPublicKey = remainderEndpoint.PublicKey.Derive(remainderPosition).GetPublicKey();
var remainderCommitment = new SecretCommitmentInfo(remainder);

var slicedEvent = CreateSliceEvent(sourceSlice, new NewSlice(receiverCommitment, receiverPublicKey), new NewSlice(remainderCommitment, remainderPublicKey));
var sourceSlicePrivateKey = await _unitOfWork.WalletRepository.GetPrivateKeyForSlice(sourceSlice.Id);
var transaction = sourceSlicePrivateKey.SignRegistryTransaction(slicedEvent.CertificateId, slicedEvent);

await SendTransactionToRegistry(transaction);

var partial = new TransferPartialSliceWaitCommittedTransactionArguments
{
Expand All @@ -108,7 +142,7 @@
SourceSliceId = msg.SourceSliceId,
TransferredSliceId = msg.TransferredSliceId,
RemainderSliceId = msg.RemainderSliceId,
TransactionId = msg.Transaction.ToShaId(),
TransactionId = transaction.ToShaId(),
ExternalEndpointId = msg.ExternalEndpointId,
RequestStatusArgs = msg.RequestStatusArgs,
WalletAttributes = msg.WalletAttributes
Expand All @@ -124,7 +158,7 @@

_logger.LogInformation("Ending consumer: {Consumer} with arguments {Args}", nameof(VaultSendRegistryTransactionConsumer), nameof(TransferPartialSliceRegistryTransactionArguments));
}
catch (Exception ex)

Check warning on line 161 in src/ProjectOrigin.Vault/EventHandlers/VaultSendRegistryTransactionConsumer.cs

View workflow job for this annotation

GitHub Actions / analyse / sonar-analysis 

Either log this exception and handle it, or rethrow it with some contextual information. (https://rules.sonarsource.com/csharp/RSPEC-2139)
{
_logger.LogError(ex, "Error sending transactions to registry");
throw;
Expand All @@ -146,4 +180,62 @@
var client = new RegistryService.RegistryServiceClient(channel);
await client.SendTransactionsAsync(request);
}

private static TransferredEvent CreateTransferEvent(WalletSlice sourceSlice, IPublicKey receiverPublicKey)
{
var sliceCommitment = new PedersenCommitment.SecretCommitmentInfo((uint)sourceSlice.Quantity, sourceSlice.RandomR);

var transferredEvent = new TransferredEvent
{
CertificateId = sourceSlice.GetFederatedStreamId(),
NewOwner = new PublicKey
{
Content = ByteString.CopyFrom(receiverPublicKey.Export()),
Type = KeyType.Secp256K1
},
SourceSliceHash = ByteString.CopyFrom(SHA256.HashData(sliceCommitment.Commitment.C))
};
return transferredEvent;
}

private sealed record NewSlice(SecretCommitmentInfo ci, IPublicKey Key);

private static SlicedEvent CreateSliceEvent(WalletSlice sourceSlice, params NewSlice[] newSlices)
{
if (newSlices.Sum(s => s.ci.Message) != sourceSlice.Quantity)
throw new InvalidOperationException();

var certificateId = sourceSlice.GetFederatedStreamId();

var sourceSliceCommitment = new PedersenCommitment.SecretCommitmentInfo((uint)sourceSlice.Quantity, sourceSlice.RandomR);
var sumOfNewSlices = newSlices.Select(newSlice => newSlice.ci).Aggregate((left, right) => left + right);
var equalityProof = SecretCommitmentInfo.CreateEqualityProof(sourceSliceCommitment, sumOfNewSlices, certificateId.StreamId.Value);

var slicedEvent = new SlicedEvent
{
CertificateId = certificateId,
SumProof = ByteString.CopyFrom(equalityProof),
SourceSliceHash = ByteString.CopyFrom(SHA256.HashData(sourceSliceCommitment.Commitment.C))
};

foreach (var newSlice in newSlices)
{
var poSlice = new SlicedEvent.Types.Slice
{
NewOwner = new PublicKey
{
Type = KeyType.Secp256K1,
Content = ByteString.CopyFrom(newSlice.Key.Export())
},
Quantity = new ProjectOrigin.Electricity.V1.Commitment
{
Content = ByteString.CopyFrom(newSlice.ci.Commitment.C),
RangeProof = ByteString.CopyFrom(newSlice.ci.CreateRangeProof(certificateId.StreamId.Value))
}
};
slicedEvent.NewSlices.Add(poSlice);
}

return slicedEvent;
}
}
43 changes: 11 additions & 32 deletions src/ProjectOrigin.Vault/EventHandlers/VaultTransferFullSliceConsumer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,17 +1,14 @@
using Google.Protobuf;
using MassTransit;
using Microsoft.Extensions.Logging;
using Npgsql;
using ProjectOrigin.Electricity.V1;
using ProjectOrigin.HierarchicalDeterministicKeys.Interfaces;
using ProjectOrigin.Vault.Database;
using ProjectOrigin.Vault.Exceptions;
using ProjectOrigin.Vault.Extensions;
using ProjectOrigin.Vault.Metrics;
using ProjectOrigin.Vault.Models;
using System;
using System.Linq;
using System.Security.Cryptography;
using System.Text.Json;
using System.Threading.Tasks;

namespace ProjectOrigin.Vault.EventHandlers;
Expand All @@ -33,7 +30,6 @@
public VaultTransferFullSliceConsumer(
IUnitOfWork unitOfWork,
ILogger<VaultTransferFullSliceConsumer> logger,
IEndpointNameFormatter formatter,
ITransferMetrics transferMetrics)
{
_unitOfWork = unitOfWork;
Expand All @@ -52,9 +48,7 @@
var sourceSlice = await _unitOfWork.CertificateRepository.GetWalletSlice(msg.SourceSliceId);
var sourceEndpoint = await _unitOfWork.WalletRepository.GetWalletEndpoint(sourceSlice.WalletEndpointId);
var externalEndpoint = await _unitOfWork.WalletRepository.GetExternalEndpoint(msg.ExternalEndpointId);

var nextReceiverPosition = await _unitOfWork.WalletRepository.GetNextNumberForId(externalEndpoint.Id);
var receiverPublicKey = externalEndpoint.PublicKey.Derive(nextReceiverPosition).GetPublicKey();

var transferredSlice = new TransferredSlice
{
Expand All @@ -69,36 +63,38 @@
};
await _unitOfWork.TransferRepository.InsertTransferredSlice(transferredSlice);

_logger.LogInformation($"Registering transfer for certificateId {sourceSlice.CertificateId}");

Check warning on line 66 in src/ProjectOrigin.Vault/EventHandlers/VaultTransferFullSliceConsumer.cs

View workflow job for this annotation

GitHub Actions / analyse / sonar-analysis 

Don't use string interpolation in logging message templates. (https://rules.sonarsource.com/csharp/RSPEC-2629)

var transferredEvent = CreateTransferEvent(sourceSlice, receiverPublicKey);

var sourceSlicePrivateKey = await _unitOfWork.WalletRepository.GetPrivateKeyForSlice(sourceSlice.Id);
var transaction = sourceSlicePrivateKey.SignRegistryTransaction(transferredEvent.CertificateId, transferredEvent);
var walletAttributes = await _unitOfWork.CertificateRepository.GetWalletAttributes(sourceEndpoint.WalletId, sourceSlice.CertificateId, sourceSlice.RegistryName, msg.HashedAttributes);

_unitOfWork.Commit();

_logger.LogInformation("Ending consumer: {Consumer}, RequestId: {RequestId} ", nameof(VaultTransferFullSliceConsumer), msg.RequestStatusArgs.RequestId);

await context.Publish<TransferFullSliceRegistryTransactionArguments>(new TransferFullSliceRegistryTransactionArguments
var message = new TransferFullSliceRegistryTransactionArguments
{
Transaction = transaction,
CertificateId = sourceSlice.CertificateId,
RegistryName = sourceSlice.RegistryName,
SliceId = sourceSlice.Id,
TransferredSliceId = transferredSlice.Id,
RequestStatusArgs = msg.RequestStatusArgs,
ExternalEndpointId = externalEndpoint.Id,
WalletAttributes = walletAttributes.ToArray()
};
await _unitOfWork.OutboxMessageRepository.Create(new OutboxMessage
{
Created = DateTimeOffset.UtcNow.ToUtcTime(),
Id = Guid.NewGuid(),
MessageType = typeof(TransferFullSliceRegistryTransactionArguments).ToString(),
JsonPayload = JsonSerializer.Serialize(message)
});

_unitOfWork.Commit();
}
catch (PostgresException ex)
{
_logger.LogError(ex, "Failed to communicate with the database.");
throw new TransientException("Failed to communicate with the database.", ex);
}
catch (Exception ex)

Check warning on line 97 in src/ProjectOrigin.Vault/EventHandlers/VaultTransferFullSliceConsumer.cs

View workflow job for this annotation

GitHub Actions / analyse / sonar-analysis 

Either log this exception and handle it, or rethrow it with some contextual information. (https://rules.sonarsource.com/csharp/RSPEC-2139)

Check warning on line 97 in src/ProjectOrigin.Vault/EventHandlers/VaultTransferFullSliceConsumer.cs

View workflow job for this annotation

GitHub Actions / analyse / sonar-analysis 

Either log this exception and handle it, or rethrow it with some contextual information. (https://rules.sonarsource.com/csharp/RSPEC-2139)
{
_unitOfWork.Rollback();
_logger.LogError(ex, "Error sending full slice transfer transactions to registry");
Expand All @@ -108,21 +104,4 @@
throw;
}
}

private static TransferredEvent CreateTransferEvent(WalletSlice sourceSlice, IPublicKey receiverPublicKey)
{
var sliceCommitment = new PedersenCommitment.SecretCommitmentInfo((uint)sourceSlice.Quantity, sourceSlice.RandomR);

var transferredEvent = new TransferredEvent
{
CertificateId = sourceSlice.GetFederatedStreamId(),
NewOwner = new PublicKey
{
Content = ByteString.CopyFrom(receiverPublicKey.Export()),
Type = KeyType.Secp256K1
},
SourceSliceHash = ByteString.CopyFrom(SHA256.HashData(sliceCommitment.Commitment.C))
};
return transferredEvent;
}
}
71 changes: 15 additions & 56 deletions src/ProjectOrigin.Vault/EventHandlers/VaultTransferPartialSliceConsumer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,6 @@
using Google.Protobuf;
using MassTransit;
using Microsoft.Extensions.Logging;
using Npgsql;
using ProjectOrigin.Electricity.V1;
using ProjectOrigin.HierarchicalDeterministicKeys.Interfaces;
using ProjectOrigin.PedersenCommitment;
using ProjectOrigin.Vault.Database;
using ProjectOrigin.Vault.Exceptions;
Expand All @@ -12,7 +9,7 @@
using ProjectOrigin.Vault.Models;
using System;
using System.Linq;
using System.Security.Cryptography;
using System.Text.Json;
using System.Threading.Tasks;

namespace ProjectOrigin.Vault.EventHandlers;
Expand Down Expand Up @@ -58,7 +55,6 @@

var receiverEndpoints = await _unitOfWork.WalletRepository.GetExternalEndpoint(msg.ExternalEndpointId);
var receiverPosition = await _unitOfWork.WalletRepository.GetNextNumberForId(receiverEndpoints.Id);
var receiverPublicKey = receiverEndpoints.PublicKey.Derive(receiverPosition).GetPublicKey();
var receiverCommitment = new SecretCommitmentInfo(quantity);
var transferredSlice = new TransferredSlice
{
Expand All @@ -75,7 +71,6 @@

var remainderEndpoint = await _unitOfWork.WalletRepository.GetWalletRemainderEndpoint(sourceEndpoint.WalletId);
var remainderPosition = await _unitOfWork.WalletRepository.GetNextNumberForId(remainderEndpoint.Id);
var remainderPublicKey = remainderEndpoint.PublicKey.Derive(remainderPosition).GetPublicKey();
var remainderCommitment = new SecretCommitmentInfo(remainder);
var remainderSlice = new WalletSlice
{
Expand All @@ -90,34 +85,39 @@
};
await _unitOfWork.CertificateRepository.InsertWalletSlice(remainderSlice);

var slicedEvent = CreateSliceEvent(sourceSlice, new NewSlice(receiverCommitment, receiverPublicKey), new NewSlice(remainderCommitment, remainderPublicKey));
var sourceSlicePrivateKey = await _unitOfWork.WalletRepository.GetPrivateKeyForSlice(sourceSlice.Id);
var transaction = sourceSlicePrivateKey.SignRegistryTransaction(slicedEvent.CertificateId, slicedEvent);
var walletAttributes = await _unitOfWork.CertificateRepository.GetWalletAttributes(sourceEndpoint.WalletId, sourceSlice.CertificateId, sourceSlice.RegistryName, msg.HashedAttributes);

_unitOfWork.Commit();

_logger.LogInformation("Ending consumer: {Consumer}, RequestId: {RequestId} ", nameof(VaultTransferPartialSliceConsumer), msg.RequestStatusArgs.RequestId);

await context.Publish<TransferPartialSliceRegistryTransactionArguments>(new TransferPartialSliceRegistryTransactionArguments
var message = new TransferPartialSliceRegistryTransactionArguments
{
Transaction = transaction,
WalletAttributes = walletAttributes.ToArray(),
ExternalEndpointId = receiverEndpoints.Id,
TransferredSliceId = transferredSlice.Id,
CertificateId = transferredSlice.CertificateId,
RegistryName = transaction.Header.FederatedStreamId.Registry,
RegistryName = sourceSlice.RegistryName,
RemainderSliceId = remainderSlice.Id,
RequestStatusArgs = msg.RequestStatusArgs,
SourceSliceId = sourceSlice.Id
SourceSliceId = sourceSlice.Id,
Quantity = msg.Quantity
};

await _unitOfWork.OutboxMessageRepository.Create(new OutboxMessage
{
Created = DateTimeOffset.UtcNow.ToUtcTime(),
Id = Guid.NewGuid(),
MessageType = typeof(TransferPartialSliceRegistryTransactionArguments).ToString(),
JsonPayload = JsonSerializer.Serialize(message)
});

_unitOfWork.Commit();
}
catch (PostgresException ex)
{
_logger.LogError(ex, "Failed to communicate with the database.");
throw new TransientException("Failed to communicate with the database.", ex);
}
catch (Exception ex)

Check warning on line 120 in src/ProjectOrigin.Vault/EventHandlers/VaultTransferPartialSliceConsumer.cs

View workflow job for this annotation

GitHub Actions / analyse / sonar-analysis 

Either log this exception and handle it, or rethrow it with some contextual information. (https://rules.sonarsource.com/csharp/RSPEC-2139)
{
_unitOfWork.Rollback();
_logger.LogError(ex, "Error sending partial slice transfer transactions to registry");
Expand All @@ -127,45 +127,4 @@
throw;
}
}

private sealed record NewSlice(SecretCommitmentInfo ci, IPublicKey Key);

private static SlicedEvent CreateSliceEvent(WalletSlice sourceSlice, params NewSlice[] newSlices)
{
if (newSlices.Sum(s => s.ci.Message) != sourceSlice.Quantity)
throw new InvalidOperationException();

var certificateId = sourceSlice.GetFederatedStreamId();

var sourceSliceCommitment = new PedersenCommitment.SecretCommitmentInfo((uint)sourceSlice.Quantity, sourceSlice.RandomR);
var sumOfNewSlices = newSlices.Select(newSlice => newSlice.ci).Aggregate((left, right) => left + right);
var equalityProof = SecretCommitmentInfo.CreateEqualityProof(sourceSliceCommitment, sumOfNewSlices, certificateId.StreamId.Value);

var slicedEvent = new SlicedEvent
{
CertificateId = certificateId,
SumProof = ByteString.CopyFrom(equalityProof),
SourceSliceHash = ByteString.CopyFrom(SHA256.HashData(sourceSliceCommitment.Commitment.C))
};

foreach (var newSlice in newSlices)
{
var poSlice = new SlicedEvent.Types.Slice
{
NewOwner = new PublicKey
{
Type = KeyType.Secp256K1,
Content = ByteString.CopyFrom(newSlice.Key.Export())
},
Quantity = new ProjectOrigin.Electricity.V1.Commitment
{
Content = ByteString.CopyFrom(newSlice.ci.Commitment.C),
RangeProof = ByteString.CopyFrom(newSlice.ci.CreateRangeProof(certificateId.StreamId.Value))
}
};
slicedEvent.NewSlices.Add(poSlice);
}

return slicedEvent;
}
}
Loading