Merge pull request #12 from terraswap/feat/sender

feat: sender

package.json

@@ -10,7 +10,7 @@
     "collect": "node --stack_size=4096 --max-old-space-size=4096 -r ts-node/register/transpile-only -r tsconfig-paths/register src/collector/main.ts",
     "migration-gen": "node --require ts-node/register -r tsconfig-paths/register ./node_modules/typeorm/cli.js migration:generate -c migration -d src/orm/migrations -n",
     "migration-revert": "node --require ts-node/register -r tsconfig-paths/register ./node_modules/typeorm/cli.js migration:revert -c migration",
-    "migration-run": "node --require ts-node/register -r tsconfig-paths/register ./node_modules/typeorm/cli.js migration:run -c migration",
+    "migration-run": "node --require ts-node/register -r tsconfig-paths/register ./node_modules/typeorm/cli.js migration:run --transaction each -c migration",
     "correction": "node --stack_size=4096 --max-old-space-size=4096 -r ts-node/register/transpile-only -r tsconfig-paths/register src/adjust.ts",
     "lint": "eslint '*/**/*.{js,ts,tsx}' --fix",
     "test": "npm test",

src/collector/indexer/common.ts

@@ -24,7 +24,7 @@ export async function getTokenPriceAsUST(
   if (isClassic) {
     return await _classicTokenPrice(manager, token, timestamp, exchangeRate)
   }
-  return await _terra2TokenPrice(manager, token, timestamp)
+  return await _mainnetTokenPrice(manager, token, timestamp)
 }
 
 async function _classicTokenPrice(manager: EntityManager,
@@ -98,7 +98,7 @@ async function _classicTokenPrice(manager: EntityManager,
   return price[token]
 }
 
-async function _terra2TokenPrice(manager: EntityManager,
+async function _mainnetTokenPrice(manager: EntityManager,
   token: string,
   timestamp: Date,
 ): Promise<UstPrice> {

src/collector/indexer/txHistoryIndexer.ts

@@ -8,6 +8,7 @@ import {
   updateVolume24h,
 } from './txHistoryUpdater'
 import { ReturningLogFinderResult } from '@terra-money/log-finder'
+import { lcd } from 'lib/terra'
 
 
 export async function TxHistoryIndexer(
@@ -17,19 +18,19 @@ export async function TxHistoryIndexer(
   txHash: string,
   founds: ReturningLogFinderResult<TxHistoryTransformed>[]
 ): Promise<void> {
-  for(const logFound of founds) {
+  for (const logFound of founds) {
     const transformed = logFound.transformed
-    if (transformed){
+    if (transformed) {
       await updateTxns(timestamp, manager, transformed.pair) // +1 to txns for pair, terraswap
+      const sender = await lcd.getContractMsgSender(txHash, transformed.pair)
       if (transformed.action === 'swap') {
         await updateVolume(manager, transformed, exchangeRate) // pair entity
         await updateVolume24h(manager, transformed, timestamp, exchangeRate) // 24h entity
       } else {
         await updateLpTokenShare(Cycle.DAY, manager, transformed)
         await updateLpTokenShare(Cycle.HOUR, manager, transformed)
       }
-      await addTxHistory(manager, timestamp, txHash, transformed) // tx history entity
+      await addTxHistory(manager, timestamp, txHash, transformed, sender) // tx history entity
     }
   }
-
 }

src/collector/indexer/txHistoryUpdater.ts

@@ -64,7 +64,8 @@ export async function addTxHistory(
   manager: EntityManager,
   timestamp: string,
   txHash: string,
-  transformed: TxHistoryTransformed
+  transformed: TxHistoryTransformed,
+  sender: string,
 ): Promise<TxHistoryEntity> {
   const txHistoryRepo = manager.getRepository(TxHistoryEntity)
 
@@ -82,6 +83,7 @@ export async function addTxHistory(
     token0Amount: isRightOrder ? transformed.assets[0].amount : transformed.assets[1].amount,
     token1: isRightOrder ? transformed.assets[1].token : transformed.assets[0].token,
     token1Amount: isRightOrder ? transformed.assets[1].amount : transformed.assets[0].amount,
+    sender,
   })
 
   return txHistoryRepo.save(txHistory)

src/collector/log-finder/log-rules.ts

@@ -1,4 +1,5 @@
 import { LogFinderRule } from '@terra-money/log-finder'
+import { isClassic, isColumbus4 } from 'lib/terra'
 import { COLUMBUS_5_COSMWASM_UPDATE_HEIGHT } from 'lib/terra/consts'
 
 
@@ -230,14 +231,14 @@ const columbus4 = {
   nativeTransferRule: col4SortedNativeTransferRule,
 }
 
-const target = process.env.TERRA_CHAIN_ID?.includes("phoenix") ? phoenix : process.env.TERRA_CHAIN_ID?.includes("columbus-4") ? columbus4 : classic
+const target = !isClassic ? phoenix : isColumbus4 ? columbus4 : classic
 
 export default {
   createPairRule: target.createPairRule,
   spwRule: target.spwRule,
   nonnativeTransferRule: target.nonnativeTransferRule,
   nativeTransferRule: target.nativeTransferRule,
   isParsable: (type: string): boolean => {
-    return (type === 'transfer') || (process.env.TERRA_CHAIN_ID?.includes("columbus-4") ? type === 'from_contract' : type === 'wasm')
+    return (type === 'transfer') || (isColumbus4 ? type === 'from_contract' : type === 'wasm')
   }
 }

src/lib/error/api.ts

@@ -54,7 +54,7 @@ export function apiErrorHandler(
   return async (ctx: Koa.Context, next: Koa.Next): Promise<void> => {
     try {
       await next()
-    } catch (error) {
+    } catch (error: any) {
       if (error instanceof APIError) {
         callback(ctx, error.type, +error.code, error.message)
       } else if (error.isJoi) {

src/lib/error/index.ts

@@ -1,6 +1,7 @@
 import * as sentry from '@sentry/node'
 import * as logger from 'lib/logger'
 import { ParamsError } from '.'
+import { AxiosError } from 'axios'
 
 export function init(
   opts: {
@@ -24,12 +25,18 @@ export function init(
   })
 }
 
-export function errorHandlerWithSentry(error?: Error): void {
-  if (error) {
-    logger.error(error)
+export function errorHandlerWithSentry(err?: AxiosError | Error): void {
+  if (err) {
+    logger.error(err)
+
+    // avoid sentry error for 404
+    if ("isAxiosError" in err && err.response?.status === 404) {
+      return
+    }
+
     // do not send Params Error to sentry
-    if (!(error instanceof ParamsError)) {
-      sentry.captureException(error)
+    if (!(err instanceof ParamsError)) {
+      sentry.captureException(err)
     }
   }
 }

src/lib/terra/block.ts

@@ -13,9 +13,17 @@ const querier = {
       params: { height },
     })
 
+    if (block.status !== 200) {
+      throw new Error(`Failed to get block results for height ${height}, ${block.data}`)
+    }
+
     const blockResults = await rpc.get(`/block_results`, {
       params: { height },
     })
+
+    if (blockResults.status !== 200) {
+      throw new Error(`Failed to get block results for height ${height}, ${blockResults.data}`)
+    }
     const blockData = JSON.parse(block.data)
     const txTimestamp = blockData.result.block.header.time
 
@@ -27,13 +35,13 @@ const querier = {
       const txHashStr = hashToHex(txString)
       const txResult = blockResultsData.result?.txs_results[idx]
       let logs = []
-      if (txResult.code === 0 ) {
+      if (txResult.code === 0) {
         logs = JSON.parse(txResult.log)
       }
       txs.push({
-        height, 
+        height,
         timestamp: txTimestamp,
-        txhash:txHashStr,
+        txhash: txHashStr,
         logs
       })
     })

src/lib/terra/fcd/classic.ts

@@ -0,0 +1,57 @@
+import axios, { AxiosInstance, AxiosRequestConfig } from 'axios'
+import * as http from 'http';
+import * as https from 'https';
+import { ClassicFcdContractMsgSenderRes, fcd } from './interfaces';
+
+
+export class ClassicFcd implements fcd {
+    private url = process.env.TERRA_FCD || 'http://localhost:8080'
+    private client: AxiosInstance;
+
+    constructor(url?: string, config?: AxiosRequestConfig) {
+        if (url) {
+            this.url = url
+        }
+        const defaultConfig = {
+            baseURL: this.url,
+            httpAgent: new http.Agent({ keepAlive: true, maxTotalSockets: 5, keepAliveMsecs: 5 * 1000 }),
+            httpsAgent: new https.Agent({ keepAlive: true, maxTotalSockets: 5 }),
+            timeout: 10 * 1000,
+        }
+        this.client = axios.create({
+            ...defaultConfig,
+            ...config,
+        })
+    }
+
+
+    async getContractMsgSender(hash: string, contract: string): Promise<string> {
+        try {
+            const result = await this.client.get<ClassicFcdContractMsgSenderRes>(`${this.url}/v1/tx/${hash}`)
+            let sender: string;
+            let found = false;
+            for (let i = 0; i < result.data?.tx?.value?.msg.length && !found; i++) {
+                const msg = result.data?.tx?.value.msg[i]
+                // maybe this msg execute the contract
+                if (!found && msg["type"]?.includes("Contract")) {
+                    sender = msg.value.sender
+                }
+                // contract direct msg
+                if (msg.value.contract === contract) {
+                    sender = msg.value.sender
+                    found = true
+                }
+            }
+            return sender
+        } catch (err: any) {
+            if (err.isAxiosError && err.response?.status === 500) {
+                const res = err.response.data
+                if (res.code !== 0 && res.message?.includes('contract query failed: unknown request')) {
+                    return undefined
+                }
+            }
+            throw err
+        }
+    }
+}
+

src/lib/terra/fcd/interfaces.ts

@@ -0,0 +1,35 @@
+export interface fcd {
+  getContractMsgSender(hash: string, contract: string): Promise<string>
+}
+
+export interface ClassicFcdContractMsgSenderRes {
+  tx: {
+    value: {
+      msg: {
+        'type': string
+        value: {
+          sender: string
+          contract?: string
+        }
+      }[]
+    }
+  }
+  height: string
+  txhash: string
+}
+
+export interface MainnetFcdContractMsgSenderRes {
+  tx: {
+    body: {
+      messages: {
+        '@type': string
+        sender: string
+        contract?: string
+      }[]
+    }
+  }
+  tx_response: {
+    height: string
+    txhash: string
+  }
+}

src/lib/terra/fcd/mainnet.ts

@@ -0,0 +1,57 @@
+import axios, { AxiosInstance, AxiosRequestConfig } from 'axios'
+import * as http from 'http';
+import * as https from 'https';
+import { MainnetFcdContractMsgSenderRes, fcd } from './interfaces';
+
+
+export class MainnetFcd implements fcd {
+    private url = process.env.TERRA_FCD || 'http://localhost:8080'
+    private client: AxiosInstance;
+
+    constructor(url?: string, config?: AxiosRequestConfig) {
+        if (url) {
+            this.url = url
+        }
+        const defaultConfig = {
+            baseURL: this.url,
+            httpAgent: new http.Agent({ keepAlive: true, maxTotalSockets: 5, keepAliveMsecs: 5 * 1000 }),
+            httpsAgent: new https.Agent({ keepAlive: true, maxTotalSockets: 5 }),
+            timeout: 10 * 1000,
+        }
+        this.client = axios.create({
+            ...defaultConfig,
+            ...config,
+        })
+    }
+
+
+    async getContractMsgSender(hash: string, contract: string): Promise<string> {
+        try {
+            const result = await this.client.get<MainnetFcdContractMsgSenderRes>(`${this.url}/v1/tx/${hash}`)
+            let sender: string;
+            let found = false;
+            for (let i = 0; i < result.data?.tx?.body?.messages.length && !found; i++) {
+                const msg = result.data.tx.body.messages[i]
+                // maybe this msg execute the contract
+                if (!found && msg["@type"]?.includes("Contract")) {
+                    sender = msg.sender
+                }
+                // contract direct msg
+                if (msg.contract === contract) {
+                    sender = msg.sender
+                    found = true
+                }
+            }
+            return sender
+        } catch (err: any) {
+            if (err.isAxiosError && err.response?.status === 500) {
+                const res = err.response.data
+                if (res.code !== 0 && res.message?.includes('contract query failed: unknown request')) {
+                    return undefined
+                }
+            }
+            throw err
+        }
+    }
+}
+

src/lib/terra/index.ts

@@ -1,4 +1,6 @@
-export const isClassic = process.env.TERRA_CHAIN_ID.includes('columbus')
+export const isClassic = process.env.TERRA_CHAIN_ID?.includes('columbus')
+export const isColumbus4 = process.env.TERRA_CHAIN_ID?.includes('columbus-4')
+
 export * from './block'
 export { default as oracle } from './oracle'
 export { default as lcd } from './lcd'