itest: assert new events RPC in some send tests

We want to make sure the new receive and send events RPC notifies on the correct
events during the send process.
We don't need to assert every single case, so we just add the new
assertion to our main send test cases.

itest/addrs_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/internal/test"
 	"github.com/lightninglabs/taproot-assets/proof"
+	"github.com/lightninglabs/taproot-assets/tapfreighter"
 	"github.com/lightninglabs/taproot-assets/tappsbt"
 	"github.com/lightninglabs/taproot-assets/taprpc"
 	wrpc "github.com/lightninglabs/taproot-assets/taprpc/assetwalletrpc"
@@ -53,17 +54,18 @@ func testAddresses(t *harnessTest) {
 	for idx, a := range rpcAssets {
 		// In order to force a split, we don't try to send the full
 		// asset.
-		addr, err := secondTapd.NewAddr(ctxt, &taprpc.NewAddrRequest{
-			AssetId:      a.AssetGenesis.AssetId,
-			Amt:          a.Amount - 1,
-			AssetVersion: a.Version,
-		})
-		require.NoError(t.t, err)
+		addr, events := NewAddrWithEventStream(
+			t.t, secondTapd, &taprpc.NewAddrRequest{
+				AssetId:      a.AssetGenesis.AssetId,
+				Amt:          a.Amount - 1,
+				AssetVersion: a.Version,
+			},
+		)
 		addresses = append(addresses, addr)
 
 		AssertAddrCreated(t.t, secondTapd, a, addr)
 
-		sendResp := sendAssetsToAddr(t, t.tapd, addr)
+		sendResp, sendEvents := sendAssetsToAddr(t, t.tapd, addr)
 		sendRespJSON, err := formatProtoJSON(sendResp)
 		require.NoError(t.t, err)
 
@@ -81,6 +83,11 @@ func testAddresses(t *harnessTest) {
 
 		// Make sure we have imported and finalized all proofs.
 		AssertNonInteractiveRecvComplete(t.t, secondTapd, idx+1)
+		AssertSendEventsComplete(t.t, addr.ScriptKey, sendEvents)
+
+		// Make sure the receiver has received all events in order for
+		// the address.
+		AssertReceiveEvents(t.t, addr, events)
 
 		// Make sure the asset meta is also fetched correctly.
 		assetResp, err := secondTapd.FetchAssetMeta(
@@ -403,7 +410,7 @@ func testAddressAssetSyncer(t *harnessTest) {
 	// group lookups by Bob.
 	AssertUniverseStats(t.t, t.tapd, 4, 4, 2)
 
-	// If Alice now mints a reissuance for the second asset group, Bob
+	// If Alice now mints a re-issuance for the second asset group, Bob
 	// should successfully sync that new asset.
 	secondGroupMember := CopyRequest(issuableAssets[1])
 	secondGroupMember.Asset.NewGroupedAsset = false
@@ -432,37 +439,40 @@ func runMultiSendTest(ctxt context.Context, t *harnessTest, alice,
 
 	// In order to force a split, we don't try to send the full asset.
 	const sendAmt = 100
-	bobAddr1, err := bob.NewAddr(ctxt, &taprpc.NewAddrRequest{
-		AssetId: genInfo.AssetId,
-		Amt:     sendAmt,
-	})
-	require.NoError(t.t, err)
+	bobAddr1, bobEvents1 := NewAddrWithEventStream(
+		t.t, bob, &taprpc.NewAddrRequest{
+			AssetId: genInfo.AssetId,
+			Amt:     sendAmt,
+		},
+	)
 	AssertAddrCreated(t.t, bob, mintedAsset, bobAddr1)
 
-	bobAddr2, err := bob.NewAddr(ctxt, &taprpc.NewAddrRequest{
-		AssetId: genInfo.AssetId,
-		Amt:     sendAmt,
-	})
-	require.NoError(t.t, err)
+	bobAddr2, bobEvents2 := NewAddrWithEventStream(
+		t.t, bob, &taprpc.NewAddrRequest{
+			AssetId: genInfo.AssetId,
+			Amt:     sendAmt,
+		})
 	AssertAddrCreated(t.t, bob, mintedAsset, bobAddr2)
 
 	// To test that Alice can also receive to multiple addresses in a single
 	// transaction as well, we also add two addresses for her.
-	aliceAddr1, err := alice.NewAddr(ctxt, &taprpc.NewAddrRequest{
-		AssetId: genInfo.AssetId,
-		Amt:     sendAmt,
-	})
-	require.NoError(t.t, err)
+	aliceAddr1, aliceEvents1 := NewAddrWithEventStream(
+		t.t, alice, &taprpc.NewAddrRequest{
+			AssetId: genInfo.AssetId,
+			Amt:     sendAmt,
+		},
+	)
 	AssertAddrCreated(t.t, alice, mintedAsset, aliceAddr1)
 
-	aliceAddr2, err := alice.NewAddr(ctxt, &taprpc.NewAddrRequest{
-		AssetId: genInfo.AssetId,
-		Amt:     sendAmt,
-	})
-	require.NoError(t.t, err)
+	aliceAddr2, aliceEvents2 := NewAddrWithEventStream(
+		t.t, alice, &taprpc.NewAddrRequest{
+			AssetId: genInfo.AssetId,
+			Amt:     sendAmt,
+		},
+	)
 	AssertAddrCreated(t.t, alice, mintedAsset, aliceAddr2)
 
-	sendResp := sendAssetsToAddr(
+	sendResp, sendEvents := sendAssetsToAddr(
 		t, alice, bobAddr1, bobAddr2, aliceAddr1, aliceAddr2,
 	)
 	sendRespJSON, err := formatProtoJSON(sendResp)
@@ -488,6 +498,14 @@ func runMultiSendTest(ctxt context.Context, t *harnessTest, alice,
 	// Make sure we have imported and finalized all proofs.
 	AssertNonInteractiveRecvComplete(t.t, bob, numRuns*2)
 	AssertNonInteractiveRecvComplete(t.t, alice, numRuns*2)
+	AssertSendEventsComplete(t.t, bobAddr1.ScriptKey, sendEvents)
+
+	// Make sure the receivers have received all events in order for the
+	//addresses.
+	AssertReceiveEvents(t.t, bobAddr1, bobEvents1)
+	AssertReceiveEvents(t.t, bobAddr2, bobEvents2)
+	AssertReceiveEvents(t.t, aliceAddr1, aliceEvents1)
+	AssertReceiveEvents(t.t, aliceAddr2, aliceEvents2)
 
 	// Now sanity check that we can actually list the transfer.
 	const (
@@ -751,23 +769,46 @@ func sendProofUniRPC(t *harnessTest, src, dst *tapdHarness, scriptKey []byte,
 // sendAssetsToAddr spends the given input asset and sends the amount specified
 // in the address to the Taproot output derived from the address.
 func sendAssetsToAddr(t *harnessTest, sender *tapdHarness,
-	receiverAddrs ...*taprpc.Addr) *taprpc.SendAssetResponse {
+	receiverAddrs ...*taprpc.Addr) (*taprpc.SendAssetResponse,
+	*EventSubscription[*taprpc.SendEvent]) {
 
 	ctxb := context.Background()
 	ctxt, cancel := context.WithTimeout(ctxb, defaultWaitTimeout)
 	defer cancel()
 
+	require.NotEmpty(t.t, receiverAddrs)
+	scriptKey := receiverAddrs[0].ScriptKey
+
 	encodedAddrs := make([]string, len(receiverAddrs))
 	for i, addr := range receiverAddrs {
 		encodedAddrs[i] = addr.Encoded
 	}
 
+	ctxc, streamCancel := context.WithCancel(ctxb)
+	stream, err := sender.SubscribeSendEvents(
+		ctxc, &taprpc.SubscribeSendEventsRequest{
+			FilterScriptKey: scriptKey,
+		},
+	)
+	require.NoError(t.t, err)
+	sub := &EventSubscription[*taprpc.SendEvent]{
+		ClientEventStream: stream,
+		Cancel:            streamCancel,
+	}
+
 	resp, err := sender.SendAsset(ctxt, &taprpc.SendAssetRequest{
 		TapAddrs: encodedAddrs,
 	})
 	require.NoError(t.t, err)
 
-	return resp
+	// We'll get events up to the point where we broadcast the transaction.
+	AssertSendEvents(
+		t.t, scriptKey, sub,
+		tapfreighter.SendStateVirtualCommitmentSelect,
+		tapfreighter.SendStateBroadcast,
+	)
+
+	return resp, sub
 }
 
 // fundAddressSendPacket asks the wallet to fund a new virtual packet with the

itest/assertions.go

@@ -20,6 +20,8 @@ import (
 	"github.com/lightninglabs/taproot-assets/commitment"
 	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/proof"
+	"github.com/lightninglabs/taproot-assets/tapfreighter"
+	"github.com/lightninglabs/taproot-assets/tappsbt"
 	"github.com/lightninglabs/taproot-assets/taprpc"
 	wrpc "github.com/lightninglabs/taproot-assets/taprpc/assetwalletrpc"
 	"github.com/lightninglabs/taproot-assets/taprpc/mintrpc"
@@ -31,11 +33,13 @@ import (
 	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/exp/maps"
+	"google.golang.org/protobuf/proto"
 )
 
 var (
 	statusDetected  = taprpc.AddrEventStatus_ADDR_EVENT_STATUS_TRANSACTION_DETECTED
 	statusConfirmed = taprpc.AddrEventStatus_ADDR_EVENT_STATUS_TRANSACTION_CONFIRMED
+	proofReceived   = taprpc.AddrEventStatus_ADDR_EVENT_STATUS_PROOF_RECEIVED
 	statusCompleted = taprpc.AddrEventStatus_ADDR_EVENT_STATUS_COMPLETED
 )
 
@@ -766,6 +770,125 @@ func AssertAddrEventByStatus(t *testing.T, client taprpc.TaprootAssetsClient,
 	require.NoError(t, err)
 }
 
+// AssertReceiveEvents makes sure all events with incremental status are sent
+// on the stream for the given address.
+func AssertReceiveEvents(t *testing.T, addr *taprpc.Addr,
+	stream *EventSubscription[*taprpc.ReceiveEvent]) {
+
+	success := make(chan struct{})
+	timeout := time.After(defaultWaitTimeout)
+	startStatus := statusDetected
+
+	// To make sure we don't forever hang on receiving on the stream, we'll
+	// cancel it after the timeout.
+	go func() {
+		select {
+		case <-timeout:
+			stream.Cancel()
+
+		case <-success:
+		}
+	}()
+
+	for {
+		event, err := stream.Recv()
+		require.NoError(t, err, "receiving address receive event")
+
+		require.True(t, proto.Equal(event.Address, addr))
+		require.Equal(t, startStatus, event.Status)
+
+		if event.Status == statusCompleted {
+			close(success)
+			stream.Cancel()
+
+			return
+		}
+
+		startStatus++
+	}
+}
+
+// AssertSendEventsComplete makes sure the two remaining events for the given
+// script key are sent on the stream.
+func AssertSendEventsComplete(t *testing.T, scriptKey []byte,
+	stream *EventSubscription[*taprpc.SendEvent]) {
+
+	AssertSendEvents(
+		t, scriptKey, stream, tapfreighter.SendStateWaitTxConf,
+		tapfreighter.SendStateComplete,
+	)
+}
+
+// AssertSendEvents makes sure all events with incremental status are sent
+// on the stream for the given script key.
+func AssertSendEvents(t *testing.T, scriptKey []byte,
+	stream *EventSubscription[*taprpc.SendEvent], from,
+	to tapfreighter.SendState) {
+
+	success := make(chan struct{})
+	timeout := time.After(defaultWaitTimeout)
+	startStatus := from
+
+	targetScriptKey, err := btcec.ParsePubKey(scriptKey)
+	require.NoError(t, err)
+
+	sendsToKey := func(e *taprpc.SendEvent) bool {
+		for _, vPacketBytes := range e.VirtualPackets {
+			vPkt, err := tappsbt.Decode(vPacketBytes)
+			require.NoError(t, err)
+
+			for _, vOut := range vPkt.Outputs {
+				if vOut.ScriptKey.PubKey == nil {
+					continue
+				}
+
+				// This packet sends to the filtered script key,
+				// we want to include this event.
+				if vOut.ScriptKey.PubKey.IsEqual(
+					targetScriptKey,
+				) {
+
+					return true
+				}
+			}
+		}
+
+		return false
+	}
+
+	// To make sure we don't forever hang on receiving on the stream, we'll
+	// cancel it after the timeout.
+	go func() {
+		select {
+		case <-timeout:
+			stream.Cancel()
+
+		case <-success:
+		}
+	}()
+
+	for {
+		event, err := stream.Recv()
+		require.NoError(t, err, "receiving send event")
+
+		require.True(t, sendsToKey(event))
+		require.Equal(t, startStatus.String(), event.SendState)
+
+		// Fully close the stream once we definitely no longer need the
+		// stream.
+		if event.SendState == tapfreighter.SendStateComplete.String() {
+			stream.Cancel()
+		}
+
+		if event.SendState == to.String() {
+			close(success)
+			return
+		}
+
+		startStatus++
+	}
+}
+
 // ConfirmAndAssertOutboundTransfer makes sure the given outbound transfer has
 // the correct state before confirming it and then asserting the confirmed state
 // with the node.

itest/burn_test.go

@@ -173,19 +173,22 @@ func testBurnAssets(t *harnessTest) {
 	// the anchor output it was in (together with the change). So let's test
 	// that we can successfully spend the change output.
 	secondSendAmt := outputAmounts[2] - burnAmt
-	fullSendAddr, err := t.tapd.NewAddr(ctxt, &taprpc.NewAddrRequest{
-		AssetId: simpleAssetGen.AssetId,
-		Amt:     secondSendAmt,
-	})
+	fullSendAddr, stream := NewAddrWithEventStream(
+		t.t, t.tapd, &taprpc.NewAddrRequest{
+			AssetId: simpleAssetGen.AssetId,
+			Amt:     secondSendAmt,
+		},
+	)
 	require.NoError(t.t, err)
 
 	AssertAddrCreated(t.t, t.tapd, simpleAsset, fullSendAddr)
-	sendResp = sendAssetsToAddr(t, t.tapd, fullSendAddr)
+	sendResp, sendEvents := sendAssetsToAddr(t, t.tapd, fullSendAddr)
 	ConfirmAndAssertOutboundTransfer(
 		t.t, minerClient, t.tapd, sendResp, simpleAssetGen.AssetId,
 		[]uint64{0, secondSendAmt}, 2, 3,
 	)
 	AssertNonInteractiveRecvComplete(t.t, t.tapd, 1)
+	AssertReceiveEvents(t.t, fullSendAddr, stream)
 
 	// Test case 3: Burn all assets of one asset ID (in this case a single
 	// collectible from the original mint TX), while there are other,
@@ -207,6 +210,7 @@ func testBurnAssets(t *harnessTest) {
 		t.t, minerClient, t.tapd, burnResp.BurnTransfer,
 		simpleCollectibleGen.AssetId, []uint64{1}, 3, 4, 1, true,
 	)
+	AssertSendEventsComplete(t.t, fullSendAddr.ScriptKey, sendEvents)
 
 	// Test case 4: Burn assets from multiple inputs. This will select the
 	// two largest inputs we have, the one over 1500 we sent above and the