ingest_vanilla.js

import Web3 from "web3";
import Knex from "knex";

import {
  loadTokenList,
  marketContract,
  borrowAndSupplyRate,
  rateToApy
} from "./compound.js";
import { subscribeNewBlocks } from "../libs/eth.js";

import config from "../config.js";

// initialize

const web3 = new Web3(`wss://mainnet.infura.io/ws/v3/${config.dev.PROJECT_ID}`);
const knex = Knex(config.dev.db);

// effect
let tokenList;
loadTokenList()
  .then(tl => (tokenList = tl))
  .catch(err => {
    throw err;
  });

// subscription
const subscription = subscribeNewBlocks(web3, async (err, blockHeader) => {
  // error handling
  if (err !== null) {
    // TODO push blockNumber onto error queue and retry at a later time
    console.error(err);
    return;
  }
  console.log(blockHeader);

  const projectName = "compound";

  // error handling
  // hack. Just skip if tokenList hasn't been loaded yet
  if (tokenList === null) {
    console.warn(
      `token list not yet loaded...skipping block ${blockHeader.number}`
    );
    return;
  }

  /*
   * written in this style, we assume that we need borrowAndSupplyRate
   * to all respond before moving on. There's no data dependency between
   * the different tokens, so we shouldn't have to do that.
   *
   * also, by using Promise.all, if one fails, they all fail.
   */

  const tokenApys = await Promise.all(
    tokenList.tokens
      // filter
      .filter(tokenRecord => tokenRecord.chainId === 1)
      // filter
      .filter(tokenRecord => tokenRecord.symbol.match(/c.*/))
      // map effects
      .map(async tokenRecord => {
        // effect
        // FIXME this is I/O what happens to other tokens if one fetch fails?
        const rates = await borrowAndSupplyRate(web3, tokenRecord.symbol);

        return {
          market_name: tokenRecord.symbol,
          borrow_rate: rates.borrowRate,
          supply_rate: rates.supplyRate
        };
      })
  );

  // map effect
  // This is just for printing. Do no processing in ingestion
  tokenApys.map(tokenApy => {
    const borrowApy = rateToApy(tokenApy.borrow_rate); // pure function
    const supplyApy = rateToApy(tokenApy.supply_rate); // pure function
    const marketName = tokenApy.market_name;
    console.log(
      `${marketName}: Supply APY = ${supplyApy.toFormat(
        18
      )}%, Borrow APY = ${borrowApy.toFormat(18)}%`
    );
  });

  // map pure
  // transform to data we want to persist
  // meader mapping?
  const apyRecords = tokenApys.map(data => {
    return {
      blocknumber: blockHeader.number,
      project_name: projectName,
      data,
      block_at: new Date(blockHeader.timestamp * 1000),
      created_at: new Date(),
      updated_at: new Date()
    };
  });

  // persist in source db
  // map effects
  await Promise.all(
    apyRecords.map(apyRecord => {
      // effect
      return knex("annual_percentage_yields")
        .insert(apyRecord)
        .catch(err => {
          // TODO push blockNumber onto error queue and retry at a later time
          console.error(err);
        });
    })
  );
  console.log("-----");
});