Skip to content
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

Add transaction-metadata topic #131

Merged
merged 5 commits into from
Jan 25, 2024
Merged

Add transaction-metadata topic #131

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
2a2982b
Implement transactions topic
jkrvivian Jan 19, 2024
47f7d4d
Add unit tests for transactions topic
jkrvivian Jan 19, 2024
bf6a8bb
Add transaction topic for specific txID
jkrvivian Jan 23, 2024
d9cbec9
Add transaction-metadata topic
jkrvivian Jan 24, 2024
44cea28
Remove `TopicTransactions` and `TopicTransaction` and fix `TopicTrans…
muXxer Jan 25, 2024
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
10 changes: 10 additions & 0 deletions pkg/mqtt/publish.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,16 @@ func (s *Server) publishBlockMetadataOnTopicsIfSubscribed(metadataFunc func() (*
)
}

func (s *Server) publishTransactionMetadataOnTopicsIfSubscribed(metadataFunc func() (*iotaapi.TransactionMetadataResponse, error), topics ...string) error {
return s.publishPayloadOnTopicsIfSubscribed(
func() (iotago.API, error) { return s.NodeBridge.APIProvider().CommittedAPI(), nil },
func() (any, error) {
return metadataFunc()
},
topics...,
)
}

func (s *Server) publishOutputIfSubscribed(ctx context.Context, output *nodebridge.Output, publishOnAllTopics bool) error {
topics := []string{GetTopicOutput(output.OutputID)}

Expand Down
31 changes: 29 additions & 2 deletions pkg/mqtt/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,7 +241,7 @@ func (s *Server) onSubscribeTopic(ctx context.Context, clientID string, topic st
go s.fetchAndPublishBlockMetadata(ctx, blockID)
}

case strings.HasPrefix(topic, "outputs/") || strings.HasPrefix(topic, "transactions/"):
case strings.HasPrefix(topic, "outputs/") || strings.HasPrefix(topic, "transactions/") || strings.HasPrefix(topic, "transaction-metadata/"):
// topicOutputs
// topicAccountOutputs
// topicAnchorOutputs
Expand All @@ -250,12 +250,16 @@ func (s *Server) onSubscribeTopic(ctx context.Context, clientID string, topic st
// topicOutputsByUnlockConditionAndAddress
// topicSpentOutputsByUnlockConditionAndAddress
// topicTransactionsIncludedBlock
// topicTransactionMetadata
s.startListenIfNeeded(ctx, GrpcListenToAcceptedTransactions, s.listenToAcceptedTransactions)
s.startListenIfNeeded(ctx, GrpcListenToLedgerUpdates, s.listenToLedgerUpdates)

if transactionID := TransactionIDFromTransactionsIncludedBlockTopic(topic); transactionID != iotago.EmptyTransactionID {
go s.fetchAndPublishTransactionInclusion(ctx, transactionID)
}
if transactionID := TransactionIDFromTransactionMetadataTopic(topic); transactionID != iotago.EmptyTransactionID {
go s.fetchAndPublishTransactionMetadata(ctx, transactionID)
}
if outputID := OutputIDFromOutputsTopic(topic); outputID != iotago.EmptyOutputID {
go s.fetchAndPublishOutput(ctx, outputID)
}
Expand Down Expand Up @@ -302,7 +306,7 @@ func (s *Server) onUnsubscribeTopic(clientID string, topic string) {
s.stopListenIfNeeded(GrpcListenToAcceptedBlocks)
s.stopListenIfNeeded(GrpcListenToConfirmedBlocks)

case strings.HasPrefix(topic, "outputs/") || strings.HasPrefix(topic, "transactions/"):
case strings.HasPrefix(topic, "outputs/") || strings.HasPrefix(topic, "transactions/") || strings.HasPrefix(topic, "transaction-metadata/"):
// topicOutputs
// topicAccountOutputs
// topicAnchorOutputs
Expand All @@ -311,6 +315,7 @@ func (s *Server) onUnsubscribeTopic(clientID string, topic string) {
// topicOutputsByUnlockConditionAndAddress
// topicSpentOutputsByUnlockConditionAndAddress
// topicTransactionsIncludedBlock
// topicTransactionMetadata
s.stopListenIfNeeded(GrpcListenToAcceptedTransactions)
s.stopListenIfNeeded(GrpcListenToLedgerUpdates)
}
Expand Down Expand Up @@ -446,6 +451,9 @@ func (s *Server) listenToAcceptedTransactions(ctx context.Context) error {
}
}

// publish the transaction metadata for this transaction in case someone subscribed to it
s.fetchAndPublishTransactionMetadata(ctx, payload.TransactionID)

// we don't return an error here, because we want to continue listening even if publishing fails once
return nil
})
Expand All @@ -463,6 +471,12 @@ func (s *Server) listenToLedgerUpdates(ctx context.Context) error {
if err := s.publishOutputIfSubscribed(ctx, created, true); err != nil {
s.LogErrorf("failed to publish created output in ledger update: %v", err)
}

// If this is the first output in a transaction (index 0), publish the transaction
// metadata for the transaction that created this output in case someone subscribed to it.
if created.OutputID.Index() == 0 {
s.fetchAndPublishTransactionMetadata(ctx, created.OutputID.TransactionID())
}
}

// we don't return an error here, because we want to continue listening even if publishing fails once
Expand Down Expand Up @@ -526,6 +540,19 @@ func (s *Server) fetchAndPublishOutput(ctx context.Context, outputID iotago.Outp
}
}

func (s *Server) fetchAndPublishTransactionMetadata(ctx context.Context, transactionID iotago.TransactionID) {
if err := s.publishTransactionMetadataOnTopicsIfSubscribed(func() (*iotaapi.TransactionMetadataResponse, error) {
resp, err := s.NodeBridge.TransactionMetadata(ctx, transactionID)
if err != nil {
return nil, ierrors.Wrapf(err, "failed to retrieve transaction metadata %s", transactionID.ToHex())
}

return resp, nil
}, GetTopicTransactionMetadata(transactionID)); err != nil {
s.LogErrorf("failed to publish transaction metadata %s: %v", transactionID.ToHex(), err)
}
}

func (s *Server) fetchAndPublishTransactionInclusion(ctx context.Context, transactionID iotago.TransactionID) {

var blockID iotago.BlockID
Expand Down
20 changes: 20 additions & 0 deletions pkg/mqtt/topics.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ const (

// single block on subscribe and changes in it's metadata (accepted, confirmed).
TopicTransactionsIncludedBlock = "transactions/" + ParameterTransactionID + "/included-block" // api.BlockWithMetadataResponse (track inclusion of a single transaction)
TopicTransactionMetadata = "transaction-metadata/" + ParameterTransactionID // api.TransactionMetadataResponse (track a specific transaction)

// single block on subscribe and changes in it's metadata (accepted, confirmed).
TopicBlockMetadata = "block-metadata/" + ParameterBlockID // api.BlockMetadataResponse (track changes to a single block)
Expand Down Expand Up @@ -102,6 +103,21 @@ func TransactionIDFromTransactionsIncludedBlockTopic(topic string) iotago.Transa
return iotago.EmptyTransactionID
}

func TransactionIDFromTransactionMetadataTopic(topic string) iotago.TransactionID {
if strings.HasPrefix(topic, "transaction-metadata/") && strings.Count(topic, "/") == 1 {
transactionIDHex := strings.Replace(topic, "transaction-metadata/", "", 1)

transactionID, err := iotago.TransactionIDFromHexString(transactionIDHex)
if err != nil || len(transactionID) != iotago.TransactionIDLength {
return iotago.EmptyTransactionID
}

return transactionID
}

return iotago.EmptyTransactionID
}

func OutputIDFromOutputsTopic(topic string) iotago.OutputID {
if strings.HasPrefix(topic, "outputs/") && strings.Count(topic, "/") == 1 {
outputIDHex := strings.Replace(topic, "outputs/", "", 1)
Expand Down Expand Up @@ -136,6 +152,10 @@ func GetTopicTransactionsIncludedBlock(transactionID iotago.TransactionID) strin
return strings.ReplaceAll(TopicTransactionsIncludedBlock, ParameterTransactionID, transactionID.ToHex())
}

func GetTopicTransactionMetadata(transactionID iotago.TransactionID) string {
return strings.ReplaceAll(TopicTransactionMetadata, ParameterTransactionID, transactionID.ToHex())
}

func GetTopicAccountOutputs(accountID iotago.AccountID, hrp iotago.NetworkPrefix) string {
return strings.ReplaceAll(TopicAccountOutputs, ParameterAccountAddress, accountID.ToAddress().Bech32(hrp))
}
Expand Down
57 changes: 57 additions & 0 deletions pkg/testsuite/mqtt_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,6 +199,53 @@ func TestMqttTopics(t *testing.T) {
}
}(),

// ok - TransactionMetadata
func() *test {
testTx := ts.NewTestTransaction(true)
transactionMetadataResponse := &api.TransactionMetadataResponse{
TransactionID: testTx.TransactionID,
TransactionState: api.TransactionStateAccepted,
}

return &test{
name: "ok - Transaction - TopicTransactionMetadata",
topics: []*testTopic{
{
topic: mqtt.GetTopicTransactionMetadata(testTx.TransactionID),
isPollingTarget: true,
isEventTarget: true,
},
},
topicsIgnore: []string{
mqtt.GetTopicOutput(testTx.ConsumedOutputID),
mqtt.GetTopicOutput(testTx.OutputID),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAny, testTx.OwnerAddress, ts.API().ProtocolParameters().Bech32HRP()),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAddress, testTx.OwnerAddress, ts.API().ProtocolParameters().Bech32HRP()),
mqtt.GetTopicTransactionsIncludedBlock(testTx.TransactionID),
},
jsonTarget: lo.PanicOnErr(ts.API().JSONEncode(transactionMetadataResponse)),
rawTarget: lo.PanicOnErr(ts.API().Encode(transactionMetadataResponse)),
preSubscribeFunc: func() {
ts.MockAddTransactionMetadata(transactionMetadataResponse.TransactionID, transactionMetadataResponse)
},
postSubscribeFunc: func() {
ts.ReceiveAcceptedTransaction(&nodebridge.AcceptedTransaction{
API: ts.API(),
Slot: testTx.BlockID.Slot(),
TransactionID: testTx.TransactionID,
// the consumed input
Consumed: []*nodebridge.Output{
ts.NewSpentNodeBridgeOutputFromTransaction(tpkg.RandBlockID(), testTx.ConsumedOutputCreationTransaction, testTx.BlockID.Slot(), testTx.TransactionID),
},
// the created output
Created: []*nodebridge.Output{
ts.NewNodeBridgeOutputFromOutputWithMetadata(testTx.OutputWithMetadataResponse),
},
})
},
}
}(),

// ok - Basic block with transaction and tagged data payload
func() *test {
testTx := ts.NewTestTransaction(true, tpkg.WithTxEssencePayload(
Expand Down Expand Up @@ -267,6 +314,7 @@ func TestMqttTopics(t *testing.T) {
mqtt.GetTopicOutput(testTx.OutputID),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAny, testTx.OwnerAddress, ts.API().ProtocolParameters().Bech32HRP()),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAddress, testTx.OwnerAddress, ts.API().ProtocolParameters().Bech32HRP()),
mqtt.GetTopicTransactionMetadata(testTx.TransactionID),
},
jsonTarget: lo.PanicOnErr(ts.API().JSONEncode(testTx.Block)),
rawTarget: lo.PanicOnErr(ts.API().Encode(testTx.Block)),
Expand Down Expand Up @@ -414,6 +462,7 @@ func TestMqttTopics(t *testing.T) {
},
topicsIgnore: []string{
mqtt.GetTopicTransactionsIncludedBlock(testTx.TransactionID),
mqtt.GetTopicTransactionMetadata(testTx.TransactionID),
mqtt.GetTopicOutput(testTx.ConsumedOutputID),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAny, testTx.OwnerAddress, ts.API().ProtocolParameters().Bech32HRP()),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAddress, testTx.OwnerAddress, ts.API().ProtocolParameters().Bech32HRP()),
Expand Down Expand Up @@ -460,6 +509,7 @@ func TestMqttTopics(t *testing.T) {
},
topicsIgnore: []string{
mqtt.GetTopicTransactionsIncludedBlock(testTx.TransactionID),
mqtt.GetTopicTransactionMetadata(testTx.TransactionID),
mqtt.GetTopicOutput(testTx.ConsumedOutputID),
mqtt.GetTopicOutput(testTx.OutputID),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAny, testTx.OwnerAddress, ts.API().ProtocolParameters().Bech32HRP()),
Expand Down Expand Up @@ -503,6 +553,7 @@ func TestMqttTopics(t *testing.T) {
},
topicsIgnore: []string{
mqtt.GetTopicTransactionsIncludedBlock(testTx.TransactionID),
mqtt.GetTopicTransactionMetadata(testTx.TransactionID),
mqtt.GetTopicOutput(testTx.ConsumedOutputID),
mqtt.GetTopicOutput(testTx.OutputID),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAny, testTx.OwnerAddress, ts.API().ProtocolParameters().Bech32HRP()),
Expand Down Expand Up @@ -548,6 +599,7 @@ func TestMqttTopics(t *testing.T) {
},
topicsIgnore: []string{
mqtt.GetTopicTransactionsIncludedBlock(testTx.TransactionID),
mqtt.GetTopicTransactionMetadata(testTx.TransactionID),
mqtt.GetTopicOutput(testTx.ConsumedOutputID),
mqtt.GetTopicOutput(testTx.OutputID),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAny, testTx.OwnerAddress, ts.API().ProtocolParameters().Bech32HRP()),
Expand Down Expand Up @@ -592,6 +644,7 @@ func TestMqttTopics(t *testing.T) {
},
topicsIgnore: []string{
mqtt.GetTopicTransactionsIncludedBlock(testTx.TransactionID),
mqtt.GetTopicTransactionMetadata(testTx.TransactionID),
mqtt.GetTopicOutput(testTx.ConsumedOutputID),
mqtt.GetTopicOutput(testTx.OutputID),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAny, testTx.OwnerAddress, ts.API().ProtocolParameters().Bech32HRP()),
Expand Down Expand Up @@ -635,6 +688,7 @@ func TestMqttTopics(t *testing.T) {
},
topicsIgnore: []string{
mqtt.GetTopicTransactionsIncludedBlock(testTx.TransactionID),
mqtt.GetTopicTransactionMetadata(testTx.TransactionID),
mqtt.GetTopicOutput(testTx.ConsumedOutputID),
mqtt.GetTopicOutput(testTx.OutputID),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAny, testTx.OwnerAddress, ts.API().ProtocolParameters().Bech32HRP()),
Expand Down Expand Up @@ -719,6 +773,7 @@ func TestMqttTopics(t *testing.T) {
},
topicsIgnore: []string{
mqtt.GetTopicTransactionsIncludedBlock(testTx.TransactionID),
mqtt.GetTopicTransactionMetadata(testTx.TransactionID),
mqtt.GetTopicOutput(testTx.ConsumedOutputID),
mqtt.GetTopicOutput(testTx.OutputID),
mqtt.GetTopicOutputsByUnlockConditionAndAddress(mqtt.UnlockConditionAny, testTx.SenderAddress, ts.API().ProtocolParameters().Bech32HRP()),
Expand Down Expand Up @@ -784,6 +839,7 @@ func TestMqttTopics(t *testing.T) {
},
topicsIgnore: []string{
mqtt.GetTopicTransactionsIncludedBlock(testTx.TransactionID),
mqtt.GetTopicTransactionMetadata(testTx.TransactionID),
mqtt.GetTopicOutput(testTx.ConsumedOutputID),
mqtt.GetTopicOutput(testTx.OutputID),
mqtt.GetTopicAnchorOutputs(anchorOutput.AnchorID, ts.API().ProtocolParameters().Bech32HRP()),
Expand Down Expand Up @@ -835,6 +891,7 @@ func TestMqttTopics(t *testing.T) {
},
topicsIgnore: []string{
mqtt.GetTopicTransactionsIncludedBlock(testTx.TransactionID),
mqtt.GetTopicTransactionMetadata(testTx.TransactionID),
mqtt.GetTopicOutput(testTx.ConsumedOutputID),
mqtt.GetTopicOutput(testTx.OutputID),
mqtt.GetTopicFoundryOutputs(lo.PanicOnErr(foundryOutput.FoundryID())),
Expand Down
26 changes: 18 additions & 8 deletions pkg/testsuite/nodebridge_mock.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,9 +27,10 @@ type MockedNodeBridge struct {
mockedLatestCommitment *nodebridge.Commitment
mockedLatestFinalizedCommitment *nodebridge.Commitment

mockedBlocks map[iotago.BlockID]*iotago.Block
mockedBlockMetadata map[iotago.BlockID]*iotaapi.BlockMetadataResponse
mockedOutputs map[iotago.OutputID]*nodebridge.Output
mockedBlocks map[iotago.BlockID]*iotago.Block
mockedBlockMetadata map[iotago.BlockID]*iotaapi.BlockMetadataResponse
mockedTransactionMetadata map[iotago.TransactionID]*iotaapi.TransactionMetadataResponse
mockedOutputs map[iotago.OutputID]*nodebridge.Output

mockedStreamListenToBlocks *MockedStream[MockedBlock]
mockedStreamListenToAcceptedBlocks *MockedStream[inx.BlockMetadata]
Expand All @@ -50,10 +51,11 @@ func NewMockedNodeBridge(t *testing.T, api iotago.API) *MockedNodeBridge {
LatestCommitmentChanged: event.New1[*nodebridge.Commitment](),
LatestFinalizedCommitmentChanged: event.New1[*nodebridge.Commitment](),
},
apiProvider: iotago.SingleVersionProvider(api),
mockedBlocks: make(map[iotago.BlockID]*iotago.Block),
mockedBlockMetadata: make(map[iotago.BlockID]*iotaapi.BlockMetadataResponse),
mockedOutputs: make(map[iotago.OutputID]*nodebridge.Output),
apiProvider: iotago.SingleVersionProvider(api),
mockedBlocks: make(map[iotago.BlockID]*iotago.Block),
mockedBlockMetadata: make(map[iotago.BlockID]*iotaapi.BlockMetadataResponse),
mockedTransactionMetadata: make(map[iotago.TransactionID]*iotaapi.TransactionMetadataResponse),
mockedOutputs: make(map[iotago.OutputID]*nodebridge.Output),
}
}

Expand Down Expand Up @@ -200,7 +202,11 @@ func (m *MockedNodeBridge) ListenToConfirmedBlocks(ctx context.Context, consumer

// TransactionMetadata returns the transaction metadata for the given transaction ID.
func (m *MockedNodeBridge) TransactionMetadata(ctx context.Context, transactionID iotago.TransactionID) (*iotaapi.TransactionMetadataResponse, error) {
panic("not implemented")
if transactionMetadata, ok := m.mockedTransactionMetadata[transactionID]; ok {
return transactionMetadata, nil
}

return nil, status.Errorf(codes.NotFound, "metadata for transaction %s not found", transactionID.ToHex())
}

func (m *MockedNodeBridge) Output(ctx context.Context, outputID iotago.OutputID) (*nodebridge.Output, error) {
Expand Down Expand Up @@ -346,6 +352,10 @@ func (m *MockedNodeBridge) MockAddBlockMetadata(blockID iotago.BlockID, blockMet
m.mockedBlockMetadata[blockID] = blockMetadata
}

func (m *MockedNodeBridge) MockAddTransactionMetadata(transactionID iotago.TransactionID, transactionMetadata *iotaapi.TransactionMetadataResponse) {
m.mockedTransactionMetadata[transactionID] = transactionMetadata
}

func (m *MockedNodeBridge) MockAddOutput(outputID iotago.OutputID, output *nodebridge.Output) {
m.mockedOutputs[outputID] = output
}
Expand Down
4 changes: 4 additions & 0 deletions pkg/testsuite/testsuite.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,10 @@ func (ts *TestSuite) MockAddBlockMetadata(blockID iotago.BlockID, blockMetadata
ts.nodeBridge.MockAddBlockMetadata(blockID, blockMetadata)
}

func (ts *TestSuite) MockAddTransactionMetadata(transactionID iotago.TransactionID, transactionMetadata *api.TransactionMetadataResponse) {
ts.nodeBridge.MockAddTransactionMetadata(transactionID, transactionMetadata)
}

func (ts *TestSuite) MockAddOutput(outputID iotago.OutputID, output *nodebridge.Output) {
ts.nodeBridge.MockAddOutput(outputID, output)
}
Expand Down
Loading