From Logs to Prod ERC20 Indexer with SQD Pipes SDK & Portal I Ian Guimaraes

GM everyone this is Ian from SQD and today we're going to go over um pipes SDK what it is why you can build with it and how to deploy it so the pipes SDK is built around u the concept of uh streams and pipes where you have one stream and many uh transformation steps and finally you have a target where is the final destination of all the of the data that was transformed. Right? So in the pipe SDK we have a portal source which is fetching the data from the SQD network piping it through a set of userdefined functions to transform or parse the data and finally you have a target which can be a database or any other kind of system that you want to output data to. Um out of the box the pipes SDK support uh click house and postgris. Also you can define your own uh your own syncs your own targets.

So it's very lightweight and the whole idea behind it is that is is meant to be very flexible for every use case. Now jumping into the code uh we're going to build a indexer um using the pipes SDK that will track your C20 transfers for basically any uh token and you're going to be able to sync it to a Postgress database. So I already have here a preset project. Uh we're using Drizzle for for this example. Uh so you if you're familiar with Drizzle, we already have the full setup here.

We're we have a Drizzle config. Um we already predefined a schema. This is going to be uh the shape of the data that's going to go to our database. So we defined a few fields and uh we also already have a index.ts file.

That's where um we're going to put together our indexer. Let's start by creating a function which will be the host for our indexer. So I'm just going to call it uh year20 transfers. And here we're going to need to first create our source. So remember that the data from the pipe SDK is coming from the SQD network through portal.

So we're going to need to create a new portal source. For the EVM portal source, we're going to need to pass a few parameters. The first one is the URL of portal and the data set that we want to use for this example. Uh here we're using Ethereum mainet, but uh we're going to have in the description um and a few other places as well that you can consult how you can get all the um available networks for for portal. Now that we created our portal source, we can start manipulating the data that is coming from portal.

The pipes SDK has a set of helper functions that help you to decode the data uh faster. So you don't need to deal with like decoding uh pipe level. So what what you can do is create a EVM decoder using the create EVM decoder function obviously. So to the EVM decoder, we pass a block range which is the range that we want to listen uh for those transactions. Here we're passing um zero which means genesis.

Uh we can also define a set of contracts uh that we want to specifically listen the events for. Uh or we can just remove this field and we're going to now listen for all the events that are matching the the transfer uh event signature. And also we need to pass an object with uh keys uh representing the event uh that we want to listen. So the pipes SDK provides a few common APIs out of the box. So here you can see we have an ERC20 uh transfer event but you can also bring your own uh APIs uh of your own contracts.

Uh we're going to have some also some some links um down below so you can check it out how you can bring your own contracts. Now that we have uh defined what kind of event that we want to listen uh we can start transforming this data to put in our database. So now that we have this whole setup we can start actually transforming the data that's coming from uh portal to the SDK. It's being already decoded through the EVM decoder. So, we don't have to do that manually.

And now we can actually uh start modifying a little bit of that data to look like this uh final schema that we want to uh sync to our database. So, that's what we're going to do in this next step. And we're going to do so using a another pipe. So, this uh pipe already has as an input uh the output of the last pipe which took the raw data and transformed uh into a decoded uh data. Right?

So in this function here uh we're going to have uh access to the transfer object and this transfer object holds uh all the transfer events that were seen in this batch of block that's being transmitted from uh portal to to the SDK and already being decoded by the EVM decoder. So we can uh map it through all those events and reshape those to look like this. So now as you can see we took all the transfer events and we map those fields inside the transfer events to match the interface that we are expecting. Uh so each event you you can um grab the the block number you can you can have um you have access to the raw event itself. So you can uh grab the log index the transaction hash and also if you want to just um get access uh straight away to all the event parameters uh you can do uh the t.

event here and you have access to all the arguments used as parameter to to that event. Uh so now the last step is just to sync the data to our database. In order to sync our data to postgris now we need to use a different function. We've been using pipe uh to transform the data. But now we need uh pipe 2 which indicates uh that's the last step of the transformation.

And inside pipe 2 uh we need to define a target. A target is an interface that allows the pipeline to output the data to any given system. And here as mentioned we already have support for postgris and click house out of the box. So we can use the function already provided by the SDK to create a new uh postgress target. As mentioned before for postgress target we're using drizzle so we need to provide the function with a few parameters once again uh one of them is a drizzle instance of postgris the other is um an array of tables so this is very important uh if you care about uh fork handling so the SDK will handle the forks internally for postgris so you need to pass every single table that is modified inside the uh on data function so if you had more tables here you need to pass all those tables to make sure that if a uh reorg happens in the chain, it's properly handled uh internally.

And of course, we need the on data function. So every time we receive a new batch of data from EVM portal source and u it's piped through all those um transformations that we have above, we're going to need to u tell where this data should go. And in this case, uh we're inserting into the transfer table because that's the only table that we have. Um and that's about it. Now we can just um run the pipeline and see if everything's working.

Before starting the pipeline, let's just run uh drizzle uh generate and migrate commands. So we make sure the database has all the tables. In this case, just one. Uh we're using uh neon database for that. So we're going to uh check back uh in the browser in a minute to make sure that we have all the tables.

So migrations are completed. Um we can check it out here if we have the tables. So yeah, the table the transfer table is here. Um now we can safely store the pipeline. And now as you can see here our indexer is progressing.

You can see um the block numbers here um the current tip of the chain and the blocks the the block range that is being um currently processed. You have the number of blocks here the amount of requests that failed or uh were successful. And if we go to uh neon interface here we can just refresh and see our data um that's coming straight from our indexer. So the pipes SDK also ships with a uh UI web interface that you can use to track the progress of your indexers. So what you need to do to enable it is to um pass a new metrics collector here u by calling this function that is also provided um by the SDK and you need to restart the server uh to make sure that the metrics server is actually collecting the metrics.

And once the server is started, we can run the pmpp pmppm dlx command um passing the the pipe package. And there we go. Uh now we can see the progress of our indexer. We also have a bunch of uh metrics collector. You can uh do performance uh metrics in your indexer.

We have ways to track that internally. So I recommend you check our documentation in order to know more about that. Uh so yeah, that's the the pipe UI. Now let's deploy our pipeline. As the SDK was built to be very flexible, we also have a lot of options when it comes to deployment, right?

Everything under the sun that you can use to to run some um Node.js continuously. So you can build a Docker image out of this. You can use um a platform as a service such as Railway or many others. Um you can also just deploy it in your bare metal server and just um be on with it.

So you have a lot of options, right? Um in this example specifically um we have a Docker file here. if you want to use um to deploy a docker image somewhere or we can also use a platform as a service. So the only um caveat in the platform as a service um deployment model is that you just need to make sure that uh the service provider will continuously run your um your app and not um scale it down when it's not being used. So there's there's a few providers that does that and we're going to see one in a in a minute.

blank is one of the platforms as a service that I just mentioned uh that supports um continuously running applications. So you don't need to worry about the platform scaling it down your uh your service and then your indexers just halts. Um so after you log in into Northflank you can just like create a new project. We're going to go through it u very quickly. um you can create a new project and deploy your your your indexer based on a either public um GitHub link or just you can um log in with your GitHub account and connect the repositories business as usual.

So here we're going to create a new service um in order to deploy our indexer. And now we just need to um select one of the repositories of the account connected or you can just like if the indexer is public the repository you can just like paste the link here. Here you can choose between docker file and build pack because the uh the repository contains the docker file in it or you can just like go with build pack and it's going to use like u the default uh node.js build build process. Here we can safely remove the HTTP um port that it has by default because if we're not exposing any metrics provider um from the SDK we don't need it.

But in case we um [snorts] you need to expose your metrics provider then you want to use um the same port as the SDK is using which is like 9,000. And now we can just create a service once we create our project. Let's just make sure to have the environment variable set for the connection string of Postgress. Um we're again using Neon here. So, we're going to create a new environment variable for that.

Now, we can click update and restart to make sure um our indexer is running with the latest uh variable. It's actually connected to start, but uh yeah, so let's just update and restart. Now that the indexer is rebuilt and restarted, we can see we already processing blocks um from portal going down through all the transformations that we defined and finally syncing the data to uh neon in this case. So yeah, those are the logs uh the runtime logs and you can also check the neon console here. We already have data coming in.

You can see here that we have um two other tables. This sync table make sure that if the indexer stops it can't resumes and it keeps track of like the syncing process of the indexer. We also have transfer snapshots which is responsible for tracking um snapshots if we need to roll back um the state well and that's it. Uh our indexer is running in production pushing data to a database. So now you can just like build stuff on top of that, consume this data, build APIs and other applications or of course like always use different contracts um to mix and match and create new data sets.

So yeah, very excited to see what you guys are going to come come up with um for the ETH global hackathon and of course outside of it. So yeah, peace.