Added Indexer Streams for SubscribeReceipts, SubscribeEvents, SubscribeBalanceUpdates (#252)

* Terminal started adding SubscribeEvents, SubscribeReceipts, SubscribeBalanceUpdates to Indexer

* added summaries for indexer event functions

* added tests for indexer event streams

* fixed DownloadHandlerStream, refined test logs

* version increment

* logic to abort all running streams

* added EventDecoded data type

* Added end-to-end tests that confirm we receive the events

---------

Co-authored-by: Quinn Purdy <[email protected]>

Author: andygruening

Assets/SequenceSDK/Indexer/Tests/ChainIndexerTests.cs

@@ -5,9 +5,14 @@
 using System.Threading.Tasks;
 using NUnit.Framework;
 using Sequence;
+using Sequence.Contracts;
+using Sequence.EmbeddedWallet;
+using Sequence.EmbeddedWallet.Tests;
 using Sequence.Utils;
 using UnityEngine;
 using UnityEngine.TestTools;
+using EOAWallet = Sequence.Wallet.EOAWallet;
+using Transaction = System.Transactions.Transaction;
 
 namespace Sequence.Indexer.Tests
 {
@@ -299,5 +304,217 @@ public async Task TestRequestDecodingErrorHandling(bool logError)
 
             Assert.IsTrue(errorEventFired);
         }
+
+        [TestCase]
+        public void TestSubscribeReceipts()
+        {
+            var indexer = new ChainIndexer(Chain.TestnetArbitrumSepolia);
+            var streamOptions = new WebRPCStreamOptions<SubscribeReceiptsReturn>(
+                OnSubscribeReceiptsMessageReceived,
+                OnWebRPCErrorReceived);
+
+            var filter = new TransactionFilter
+            {
+                contractAddress = "0x4ab3b16e9d3328f6d8025e71cefc64305ae4fe9c"
+            };
+
+            indexer.SubscribeReceipts(new SubscribeReceiptsArgs(filter), streamOptions);
+        }
+
+        [Test]
+        public async Task TestReceiptSubscriptionReceived()
+        {
+            Chain chain = Chain.ArbitrumNova;
+            bool receiptReceived = false;
+            IIndexer indexer = new ChainIndexer(chain);
+            ERC1155 universallyMintable = new ERC1155("0x0ee3af1874789245467e7482f042ced9c5171073");
+            var streamOptions = new WebRPCStreamOptions<SubscribeReceiptsReturn>(
+                (@return =>
+                {
+                    Assert.IsNotNull(@return);
+                    Assert.IsNotNull(@return.receipt);
+                    receiptReceived = true;
+                }),
+                OnWebRPCErrorReceived);
+
+
+            var filter = new TransactionFilter()
+            {
+                contractAddress = universallyMintable.Contract.GetAddress()
+            };
+            indexer.SubscribeReceipts(new SubscribeReceiptsArgs(filter), streamOptions);
+
+            await DoMintTransaction(chain);
+
+            while (!receiptReceived)
+            {
+                await Task.Yield();
+            }
+        }
+
+        private async Task DoMintTransaction(Chain chain)
+        {
+            var tcs = new TaskCompletionSource<bool>();
+            EndToEndTestHarness testHarness = new EndToEndTestHarness();
+            EOAWallet toWallet = new EOAWallet();
+            string toAddress = toWallet.GetAddress().Value;
+            
+            testHarness.Login(async wallet =>
+            {
+                try
+                {
+                    ERC1155 erc1155 = new ERC1155("0x0ee3af1874789245467e7482f042ced9c5171073");
+
+                    TransactionReturn transactionReturn = await wallet.SendTransaction(chain,
+                        new EmbeddedWallet.Transaction[]
+                        {
+                            new RawTransaction(erc1155.Mint(toAddress, 1, 1)),
+                        });
+                    Assert.IsNotNull(transactionReturn);
+                    Assert.IsTrue(transactionReturn is SuccessfulTransactionReturn);
+                    
+                    tcs.TrySetResult(true);
+                }
+                catch (System.Exception e)
+                {
+                    tcs.TrySetException(e);
+                }
+            }, (error, method, email, methods) =>
+            {
+                tcs.TrySetException(new Exception(error));
+            });
+
+            await tcs.Task;
+        }
+        
+        [TestCase]
+        public void TestSubscribeEvents()
+        {
+            var indexer = new ChainIndexer(Chain.TestnetArbitrumSepolia);
+            var streamOptions = new WebRPCStreamOptions<SubscribeEventsReturn>(
+                OnSubscribeEventsMessageReceived,
+                OnWebRPCErrorReceived);
+
+            var eventFilter = new EventFilter
+            {
+                accounts = Array.Empty<string>(),
+                contractAddresses = new[] {"0x4ab3b16e9d3328f6d8025e71cefc64305ae4fe9c"},
+                tokenIDs = new[] {"0"},
+                events = new[] {"Transfer(address from, address to, uint256 value)"}
+            };
+            
+            indexer.SubscribeEvents(new SubscribeEventsArgs(eventFilter), streamOptions);
+        }
+
+        [Test]
+        public async Task TestEventSubscriptionReceived()
+        {
+            Chain chain = Chain.ArbitrumNova;
+            bool eventReceived = false;
+            IIndexer indexer = new ChainIndexer(chain);
+            ERC1155 universallyMintable = new ERC1155("0x0ee3af1874789245467e7482f042ced9c5171073");
+            var streamOptions = new WebRPCStreamOptions<SubscribeEventsReturn>(
+                (@return =>
+                {
+                    Assert.IsNotNull(@return);
+                    Assert.IsNotNull(@return.log);
+                    Assert.AreEqual(universallyMintable.Contract.GetAddress().Value, @return.log.contractAddress);
+                    eventReceived = true;
+                }),
+                OnWebRPCErrorReceived);
+
+
+            var eventFilter = new EventFilter
+            {
+                accounts = Array.Empty<string>(),
+                contractAddresses = new[] {universallyMintable.Contract.GetAddress().Value},
+                tokenIDs = new[] {"1"},
+                events = new[] {"TransferSingle(address indexed operator, address indexed from, address indexed to, uint256 id, uint256 value)"}
+            };
+            
+            indexer.SubscribeEvents(new SubscribeEventsArgs(eventFilter), streamOptions);
+
+            await DoMintTransaction(chain);
+
+            while (!eventReceived)
+            {
+                await Task.Yield();
+            }
+        }
+        
+        [TestCase]
+        public void TestSubscribeBalanceUpdates()
+        {
+            var indexer = new ChainIndexer(Chain.TestnetArbitrumSepolia);
+            var streamOptions = new WebRPCStreamOptions<SubscribeBalanceUpdatesReturn>(
+                OnSubscribeEventsMessageReceived,
+                OnWebRPCErrorReceived);
+
+            var contractAddress = "0x4ab3b16e9d3328f6d8025e71cefc64305ae4fe9c";
+            indexer.SubscribeBalanceUpdates(new SubscribeBalanceUpdatesArgs(contractAddress), streamOptions);
+        }
+
+        [Test]
+        public async Task TestBalanceUpdateSubscriptionReceived()
+        {
+            Chain chain = Chain.ArbitrumNova;
+            bool eventReceived = false;
+            IIndexer indexer = new ChainIndexer(chain);
+            var streamOptions = new WebRPCStreamOptions<SubscribeBalanceUpdatesReturn>(
+                (@return =>
+                {
+                    Assert.IsNotNull(@return);
+                    Assert.IsNotNull(@return.balance);
+                    if (!@return.balance.accountAddress.IsZeroAddress())
+                    {
+                        Assert.AreEqual(BigInteger.One, @return.balance.balance);
+                    }
+                    eventReceived = true;
+                }),
+                OnWebRPCErrorReceived);
+
+            ERC1155 universallyMintable = new ERC1155("0x0ee3af1874789245467e7482f042ced9c5171073");
+            
+            indexer.SubscribeBalanceUpdates(new SubscribeBalanceUpdatesArgs(universallyMintable.Contract.GetAddress()), streamOptions);
+
+            await DoMintTransaction(chain);
+
+            while (!eventReceived)
+            {
+                await Task.Yield();
+            }
+        }
+
+        [TestCase]
+        public void TestAbortStreams()
+        {
+            new ChainIndexer(Chain.TestnetArbitrumSepolia).AbortStreams();
+        }
+
+        [TearDown]
+        public void UnsubscribeFromEvents()
+        {
+            new ChainIndexer(Chain.ArbitrumNova).AbortStreams();
+        }
+        
+        private void OnSubscribeReceiptsMessageReceived(SubscribeReceiptsReturn @event)
+        {
+            Debug.Log($"Receipt Event Received - hash: {@event.receipt.txnHash}");
+        }
+
+        private void OnSubscribeEventsMessageReceived(SubscribeEventsReturn @event)
+        {
+            Debug.Log($"Contract Event Received - {@event.log.type} {@event.log.contractType}: {@event.log.txnHash}");
+        }
+        
+        private void OnSubscribeEventsMessageReceived(SubscribeBalanceUpdatesReturn @event)
+        {
+            Debug.Log($"Balance Update Received - {@event.balance.accountAddress} owns {@event.balance.balance}");
+        }
+
+        private void OnWebRPCErrorReceived(WebRPCError error)
+        {
+            Debug.LogError($"OnWebRPCErrorReceived: {error.msg}");
+        }
     }
 }

Assets/SequenceSDK/Indexer/Tests/MockHttpHandler.cs

@@ -26,5 +26,15 @@ public Task<string> HttpPost(string chainID, string endPoint, object args, int r
             }
             return Task.FromResult(_response);
         }
+
+        public void HttpStream<T>(string chainID, string endPoint, object args, WebRPCStreamOptions<T> options, int retries = 0)
+        {
+            throw new NotImplementedException();
+        }
+
+        public void AbortStreams()
+        {
+            throw new NotImplementedException();
+        }
     }
 }

Assets/SequenceSDK/Indexer/Tests/SequenceIndexerTests.asmdef

@@ -7,7 +7,10 @@
         "GUID:0acc523941302664db1f4e527237feb3",
         "GUID:a35e3a53d4439435f8b36ed2c6158cd8",
         "GUID:b4f9c0f8f363f439b9e337f79050f189",
-        "GUID:403077141e1554429a890cbc129df651"
+        "GUID:403077141e1554429a890cbc129df651",
+        "GUID:040286810a82b46ed9acd6d70bfbbfd4",
+        "SequenceEmbeddedWallet",
+        "SequenceAuthentication"
     ],
     "includePlatforms": [
         "Editor"

Packages/Sequence-Unity/Sequence/SequenceFrontend/Scripts/UI/SequenceSampleUI.cs

@@ -1,3 +1,4 @@
+using System;
 using Sequence.EmbeddedWallet;
 using UnityEngine;
 

Packages/Sequence-Unity/Sequence/SequenceSDK/Ethereum/Address/Address.cs

@@ -82,4 +82,19 @@ public override bool CanConvert(Type objectType)
             return true;
         }
     }
+
+    public static class AddressExtensions
+    {
+        public static bool IsZeroAddress(this Address address)
+        {
+            return IsZeroAddress(address.Value);
+        }
+
+        public static bool IsZeroAddress(this string address)
+        {
+            string toCheck = address.WithoutHexPrefix();
+            toCheck = toCheck.Replace("0", "");
+            return toCheck.Length == 0;
+        }
+    }
 }

Packages/Sequence-Unity/Sequence/SequenceSDK/Ethereum/Provider/DataTypes/TransactionReceipt.cs

@@ -6,6 +6,7 @@ namespace Sequence
     [System.Serializable]
     public class TransactionReceipt
     {
+        public string txnHash;
         public string transactionHash;
         public string transactionIndex;
         public string blockHash;
@@ -22,6 +23,4 @@ public class TransactionReceipt
         public string root;
         public string status;
     }
-
-   
 }

Packages/Sequence-Unity/Sequence/SequenceSDK/Indexer/ChainIndexer.cs

@@ -148,5 +148,25 @@ public Task<GetTransactionHistoryReturn> GetTransactionHistory(GetTransactionHis
         {
             return Indexer.GetTransactionHistory(ChainId, args, 0, _customHttpHandler, this);
         }
+
+        public void SubscribeReceipts(SubscribeReceiptsArgs args, WebRPCStreamOptions<SubscribeReceiptsReturn> options)
+        {
+            Indexer.SubscribeReceipts(ChainId, args, options, this);
+        }
+        
+        public void SubscribeEvents(SubscribeEventsArgs args, WebRPCStreamOptions<SubscribeEventsReturn> options)
+        {
+            Indexer.SubscribeEvents(ChainId, args, options, this);
+        }
+        
+        public void SubscribeBalanceUpdates(SubscribeBalanceUpdatesArgs args, WebRPCStreamOptions<SubscribeBalanceUpdatesReturn> options)
+        {
+            Indexer.SubscribeBalanceUpdates(ChainId, args, options, this);
+        }
+
+        public void AbortStreams()
+        {
+            Indexer.AbortStreams(this);
+        }
     }
 }

Packages/Sequence-Unity/Sequence/SequenceSDK/Indexer/DataTypes/EventDecoded.cs

@@ -0,0 +1,12 @@
+namespace Sequence
+{
+    [System.Serializable]
+    public class EventDecoded
+    {
+        public string topicHash;
+        public string eventSig;
+        public string[] types;
+        public string[] names;
+        public string[] values;
+    }
+}

Packages/Sequence-Unity/Sequence/SequenceSDK/Indexer/DataTypes/EventDecoded.cs.meta

@@ -0,0 +1,11 @@
+namespace Sequence
+{
+    [System.Serializable]
+    public class EventFilter
+    {
+        public string[] events;
+        public string[] contractAddresses;
+        public string[] accounts;
+        public string[] tokenIDs;
+    }
+}