EOS Demux using PostgreSQL

I have been looking for EOS demux implementation using PostgreSQL but found none. So I thought of writing this short tutorial explaining how we can implement EOS demux using PostgreSQL. Those who don’t know about EOS Demux please have a look at the article explaining the same in a very intuitive way.  https://github.com/EOSIO/demux-js.

 

A short explanation of EOS Demux: Demux architecture allows application developers to use traditional Mongo or Postgres SQL databases in a way that means the data stored in them is still verifiable by the blockchain. This enables the best of both worlds: the flexibility and speed of traditional databases, coupled with the trust and immutable properties of a blockchain. The overall flow of an application looks as follows:

 

To use PostgreSQL as a data store we need to use massive.js. So go ahead and install massive.js using the following command

> npm i massive --save

#include <eosiolib/eosio.hpp>

#include <eosiolib/multi_index.hpp>

#include <eosiolib/types.hpp>

#include <eosiolib/print.hpp>

#include <eosiolib/system.h>

//using namespace eosio;


class blog_contract : public eosio::contract{

    public:


    blog_contract(account_name self) : eosio::contract(self),

          bloges(self, self){}


        /// @abi action

        void create(account_name author, uint64_t id, std::string email, const std::string& data){

            require_auth(author);

            blogs.emplace(author, [&](auto& new_blog) {

                new_blog.id = id;

                new_blog.author = author;

            });

        }

        //.. destroy, show and update actions .. 


    private :

        /// @abi table blog i64

        struct blog{

            uint64_t id;

            uint64_t author;

            uint64_t primary_key() const {

                return id;

            }

            EOSLIB_SERIALIZE(blog,(id)(author));

        };

        typedef eosio::multi_index<N(blog), blog> blog_table;

        blog_table blogs;

};


EOSIO_ABI( blog_contract, (create)(destroy)(show)(update) )

If you look closely at ‘create’ function of the contract, you would notice that the function’s signature contains blog data as one of the parameters but we are not storing the blog data as part of EOS’ database. The purpose of having blog data as part of the function signature is that it gets available to be stored in PostgreSQL. Whatever we mentioned as function parameters, those parameter’s data will be available to the action handler. Looking at the diagram ‘Action watcher’ keeps looking for any change in blockchain via ‘Action Reader’. Whenever there is any action happened in the Blockchain, Action watcher delegate that action to ‘Action Handler’. Then ‘Action Handler’ further calls ‘updaters’ and ‘effects’ which actually execute necessary action code. Based upon the discussion the configuration looks as follows:

const massive = require('massive');

let db;

massive({
 host: host,
 port: port,
 database: database, 
 user: user,
 password: password 
}).then(instance => {
 db = instance;
 return Promise.resolve(db);
}).catch(e => {
 console.log(e)
 console.log('error while getting massive instance')
 });

function createBlog(db, payload, blockInfo, context){
 db.blog_data.insert({
 id : payload.data.id,
 author : payload.data.author,
 data : payload.data.data, //Storing blog data
 email : payload.data.email
 }).then(new_blog => {
 console.log('blog created')
 }).catch(err => {
 console.log('error while inserting data')
 })
}

//Register blockchain Actions to which this update will be called.

const updaters = [{
 actionType: "blogger.p::create",
 updater: createProfile}
]

module.exports = updaters

const {
 readers: {
 eos: { NodeosActionReader }
 },
 handlers :{
 postgres : { MassiveActionHandler }
 },
 watchers: { BaseActionWatcher }
 } = require('demux-js')

const actionHandler = new MassiveActionHandler(
 updaters,
 effects,
 db
)

const actionReader = new NodeosActionReader(
 httpEndpoint,
 0, // Start at most recent blocks
)


const actionWatcher = new BaseActionWatcher(
 actionReader,
 actionHandler,
500,
)
actionWatcher.watch()

CREATE TABLE _index_state (
 id serial PRIMARY KEY,
 block_number integer NOT NULL,
 block_hash text NOT NULL,
 is_replay boolean NOT NULL
);