Add tx history endpoints

deployment/features/deployment.yaml

@@ -14,10 +14,10 @@ spec:
         app: subid-back-<BRANCH>
     spec:
       imagePullSecrets:
-        - name: dockerhub 
+        - name: dockerhub
       containers:
         - name: subid-back-<BRANCH>
-          image: <IMAGE> 
+          image: <IMAGE>
           imagePullPolicy: IfNotPresent
           ports:
             - containerPort: 3001
@@ -37,6 +37,15 @@ spec:
               value: redis-master.default
             - name: REDIS_PORT
               value: '6379'
+
+            - name: AGGREGATOR_REDIS_HOST
+              value: redis-aggregation-datasource-subquery-master.default
+            - name: AGGREGATOR_REDIS_PASSWORD
+              value: 2KK6RWECApEZJjmd
+            - name: AGGREGATOR_REDIS_PREFIX
+              value: aggregator_queue_datasource-subquery
+            - name: AGGREGATOR_REDIS_PORT
+              value: '6379'
           envFrom:
             - secretRef:
                 name: subid-back-secret

package.json

@@ -22,6 +22,7 @@
   "devDependencies": {
     "@subsocial/config": "0.7.7",
     "@types/bn.js": "^5.1.0",
+    "@types/bull": "^4.10.0",
     "@types/connect-timeout": "^0.0.34",
     "@types/cors": "^2.8.6",
     "@types/express": "^4.17.3",
@@ -45,6 +46,7 @@
     "@polkawallet/bridge": "0.1.3",
     "@subsocial/api": "0.8.14",
     "axios": "^0.26.0",
+    "bull": "^4.11.4",
     "connect-timeout": "^1.9.0",
     "cors": "^2.8.5",
     "dayjs": "^1.10.4",

src/constant/graphQlClients.ts

@@ -4,17 +4,30 @@ import { RelayChain } from '../services/crowdloan/types'
 
 export const subsocialGraphQlClient = new GraphQLClient(SUBSOCIAL_GRAPHQL_CLIENT)
 
-export const soonsocialGraphQlClient = new GraphQLClient('https://squid.subsquid.io/soonsocial/graphql')
+export const soonsocialGraphQlClient = new GraphQLClient(
+  'https://squid.subsquid.io/soonsocial/graphql'
+)
 
-export const contributionsClientByRelay: Record<RelayChain, { client: GraphQLClient; addressPrefix: number }> = {
+export const txAggregatorGraphQlClient = new GraphQLClient(
+  'https://tx-aggregation.subsocial.network/graphql'
+)
+
+export const contributionsClientByRelay: Record<
+  RelayChain,
+  { client: GraphQLClient; addressPrefix: number }
+> = {
   kusama: {
-    client: new GraphQLClient('https://squid.subsquid.io/kusama-explorer/graphql', { timeout: 4000 }),
+    client: new GraphQLClient('https://squid.subsquid.io/kusama-explorer/graphql', {
+      timeout: 4000
+    }),
     addressPrefix: 2
   },
   polkadot: {
-    client: new GraphQLClient('https://squid.subsquid.io/polkadot-explorer/graphql', { timeout: 4000 }),
+    client: new GraphQLClient('https://squid.subsquid.io/polkadot-explorer/graphql', {
+      timeout: 4000
+    }),
     addressPrefix: 0
   }
 }
 
-export const quartzClient = new GraphQLClient(QUARTZ_GRAPHQL_CLIENT)
+export const quartzClient = new GraphQLClient(QUARTZ_GRAPHQL_CLIENT)

src/routes/index.ts

@@ -18,6 +18,7 @@ import createFeesRouter from './fees'
 import { Connections } from '../connections'
 import createCreatorStakingRouter from './creatorStaking'
 import { isDef } from '@subsocial/utils';
+import createTxHistoryRouter from './txHistory'
 
 export const createRoutes = (apis: Connections) => {
   const router = Router()
@@ -79,7 +80,7 @@ export const createRoutes = (apis: Connections) => {
       const result = balances.map((balance: any) => {
         const { status, value } = balance
 
-        return status === 'fulfilled' && value
+        return status === 'fulfilled' ? value : undefined
       })
 
       res.send(result.filter(isDef))
@@ -151,6 +152,7 @@ export const createRoutes = (apis: Connections) => {
   router.use('/staking/collator', asyncErrorHandler(createCollatorStakingRouter(apis.mixedApis)))
   router.use('/staking/validator', asyncErrorHandler(createValidatorStakingRouter(apis)))
   router.use('/staking/creator', asyncErrorHandler(createCreatorStakingRouter(apis)))
+  router.use('/tx', asyncErrorHandler(createTxHistoryRouter()))
 
   router.use('/fees', asyncErrorHandler(createFeesRouter(apis.mixedApis)))
 

src/routes/txHistory/index.ts

@@ -0,0 +1,34 @@
+import { Router } from 'express'
+import { getAccountTxHistory, getAccountTxHistoryWithQueue } from '../../services/txHistory'
+
+const createTxHistoryRouter = () => {
+  const router = Router()
+
+  router.get('/history/queue', async function (req, res) {
+    const { address, pageSize, offset } = req.query
+    const txs = await getAccountTxHistoryWithQueue({
+      address: address as string,
+      pageSize: parseInt(pageSize as string),
+      offset: parseInt(offset as string),
+    })
+
+    res.send(txs)
+  })
+
+  router.get('/history', async function (req, res) {
+    const { address, pageSize, offset, networks, events } = req.query
+    const txs = await getAccountTxHistory({
+      address: address as string,
+      pageSize: parseInt(pageSize as string),
+      offset: parseInt(offset as string),
+      networks: networks as string[],
+      events: events as string[],
+    })
+
+    res.send(txs)
+  })
+
+  return router
+}
+
+export default createTxHistoryRouter

src/server.ts

@@ -1,7 +1,7 @@
 import express from 'express'
 import cors from 'cors'
 import timeout from 'connect-timeout'
-import { reqTimeoutSecs, allowedOrigins, port } from './constant/env'
+import { reqTimeoutSecs, port, allowedOrigins } from './constant/env'
 import { newLogger } from '@subsocial/utils'
 
 import { createRoutes } from './routes'
@@ -36,7 +36,7 @@ export const startHttpServer = (apis: Connections) => {
     })
   )
 
-  // For localhost testing 
+  // For localhost testing
   // app.use(
   //   cors((req, callback) => {
   //     const origin = req.method === 'GET' ? '*' : '*'

src/services/identity/getIdentityFromSquid.ts

@@ -4,7 +4,10 @@ import { identitiesInfoCache } from '.'
 import { toGenericAccountId } from '../utils'
 import { u8aToHex } from '@polkadot/util'
 import { getIdentityFromChain } from './getIdentityFromChain'
-import { ApiPromise } from '@polkadot/api';
+import { ApiPromise } from '@polkadot/api'
+import { newLogger } from '@subsocial/utils'
+
+const log = newLogger('Identity')
 
 const GET_IDENTITY = gql`
   query GetIdentity($ids: [String!]) {
@@ -94,8 +97,8 @@ export const tryToGetIdentityFromSquid = async (
 ) => {
   try {
     await getIdentityFromSquid(accounts, chain)
-  } catch {
-    console.error('Failed to get identity from squid, trying to get from chain')
+  } catch (e) {
+    log.warn('Failed to get identity from squid, trying to get from chain', e)
     await getIdentityFromChain(api, accounts, chain)
   }
 }

src/services/prices.ts

@@ -53,8 +53,6 @@ export const getPrices = async (ids: string) => {
   const forceUpdate = needUpdate && (await needUpdate())
   const cacheData = await pricesCache.get(cacheKey)
 
-  console.log('Is Empty cache data: ', !cacheData?.values || !cacheData?.values.length)
-
   if (!cacheData?.values || !cacheData?.values.length) {
     await fetchPrices(ids)
   }

src/services/txHistory/index.ts

@@ -0,0 +1,110 @@
+import { gql } from 'graphql-request'
+import { u8aToHex } from '@polkadot/util'
+import { decodeAddress, isEthereumAddress } from '@polkadot/util-crypto'
+import { getOrCreateQueue } from './queue'
+import { txAggregatorGraphQlClient } from '../../constant/graphQlClients'
+
+const ADD_QUEUE_JOB_NAME = 'REFRESH_TX_HISTORY_FOR_ACCOUNT_ON_DEMAND'
+
+const buildGetAccountTxHistoryQuery = (networks?: string[], events?: string[]) => {
+  const networkFilterValues = networks ? ', blockchainTag: $networks' : ''
+  const networkFilterParams = networks ? ', $networks: [BlockchainTag!]' : ''
+
+  const eventFilterValues = events ? ', txKind: $txKind' : ''
+  const eventFilterParams = events ? ', $txKind: [TransactionKind!]' : ''
+
+  return gql`
+    query getAccountTxHistory($address: String!, $pageSize: Int!, $offset: Int! ${networkFilterParams}${eventFilterParams}) {
+      accountTxHistory(
+        args: { where: { publicKey: $address ${networkFilterValues}${eventFilterValues} }, pageSize: $pageSize, offset: $offset }
+      ) {
+        data {
+          id
+          txKind
+          blockchainTag
+          amount
+          senderOrTargetPublicKey
+          timestamp
+          success
+          transaction {
+            transferNative {
+              extrinsicHash
+            }
+          }
+        }
+      }
+    }
+  `
+}
+
+type GetAccountTransactionsWithQueue = {
+  address: string
+  pageSize: number
+  offset: number
+}
+
+type GetAccountTransactions = GetAccountTransactionsWithQueue & {
+  networks?: string[]
+  events?: string[]
+}
+
+export const getAccountTxHistory = async ({
+  address,
+  pageSize,
+  offset,
+  networks,
+  events
+}: GetAccountTransactions) => {
+  const networkFilterValues = networks
+    ? { networks: networks.map((network) => network.toUpperCase()) }
+    : {}
+  const eventsFilterValues = events ? { txKind: events.map((event) => event.toUpperCase()) } : {}
+
+  const query = buildGetAccountTxHistoryQuery(networks, events)
+
+  const txs = await txAggregatorGraphQlClient.request(query, {
+    address,
+    pageSize,
+    offset,
+    ...networkFilterValues,
+    ...eventsFilterValues
+  })
+
+  return txs?.accountTxHistory?.data
+}
+
+export const getAccountTxHistoryWithQueue = async (props: GetAccountTransactionsWithQueue) => {
+  const txs = await getAccountTxHistory(props)
+
+  const address = props.address
+
+  const jobId = `${address}-${ADD_QUEUE_JOB_NAME}`
+  const queue = getOrCreateQueue()
+  const jobByAddress = await queue.getJob(jobId)
+
+  let actualData = false
+
+  if (jobByAddress) {
+    const jobState = await jobByAddress.getState()
+
+    if (jobState === 'completed') {
+      jobByAddress.remove()
+
+      actualData = true
+    }
+  } else {
+    const taskPayload = {
+      publicKey: isEthereumAddress(address) ? address : u8aToHex(decodeAddress(address))
+    }
+
+    queue.add(ADD_QUEUE_JOB_NAME, taskPayload, {
+      attempts: 5,
+      jobId,
+      removeOnComplete: false,
+      removeOnFail: true,
+      priority: 1
+    })
+  }
+
+  return { txs, actualData }
+}

src/services/txHistory/queue.ts

@@ -0,0 +1,30 @@
+import Queue from 'bull'
+
+let queue = null
+
+export const getOrCreateQueue = () => {
+  if (!queue) {
+    const host = process.env.AGGREGATOR_REDIS_HOST
+    const password = process.env.AGGREGATOR_REDIS_PASSWORD
+    const port = process.env.AGGREGATOR_REDIS_PORT as unknown as number
+
+    const aggregatorRedisConfig = {
+      host,
+      password,
+      port
+    }
+
+    queue = new Queue('ACCOUNT_AGGREGATION_FLOW', {
+      redis: aggregatorRedisConfig,
+      prefix: process.env.AGGREGATOR_REDIS_PREFIX,
+      settings: {
+        lockDuration: 20000, // Check for stalled jobs each 2 min
+        lockRenewTime: 10000,
+        stalledInterval: 20 * 60 * 1000,
+        maxStalledCount: 1
+      }
+    })
+  }
+
+  return queue
+}