queryEngine.ts

import { Semaphore } from 'async-mutex'
import nano from 'nano'

import { config } from './config'
import { pagination } from './dbutils'
import { banxa } from './partners/banxa'
import { bitaccess } from './partners/bitaccess'
import { bitrefill } from './partners/bitrefill'
import { bitsofgold } from './partners/bitsofgold'
import { bity } from './partners/bity'
import { changehero } from './partners/changehero'
import { changelly } from './partners/changelly'
import { changenow } from './partners/changenow'
import { exolix } from './partners/exolix'
import { foxExchange } from './partners/foxExchange'
import { gebo } from './partners/gebo'
import { godex } from './partners/godex'
import { ioniaGiftCards } from './partners/ioniagiftcard'
import { ioniaVisaRewards } from './partners/ioniavisarewards'
import { kado } from './partners/kado'
import { letsexchange } from './partners/letsexchange'
import { libertyx } from './partners/libertyx'
import { lifi } from './partners/lifi'
import { moonpay } from './partners/moonpay'
import { paybis } from './partners/paybis'
import { paytrie } from './partners/paytrie'
import { rango } from './partners/rango'
import { safello } from './partners/safello'
import { sideshift } from './partners/sideshift'
import { simplex } from './partners/simplex'
import { swapuz } from './partners/swapuz'
import { switchain } from './partners/switchain'
import { maya, thorchain } from './partners/thorchain'
import { transak } from './partners/transak'
import { wyre } from './partners/wyre'
import { xanpool } from './partners/xanpool'
import {
  asApp,
  asApps,
  asDisablePartnerQuery,
  asProgressSettings,
  DbTx,
  DisablePartnerQuery,
  ScopedLog,
  StandardTx
} from './types'
import {
  createScopedLog,
  datelog,
  promiseTimeout,
  standardizeNames
} from './util'

const nanoDb = nano(config.couchDbFullpath)

const plugins = [
  banxa,
  bitaccess,
  bitsofgold,
  bity,
  bitrefill,
  changelly,
  changenow,
  changehero,
  exolix,
  foxExchange,
  gebo,
  godex,
  ioniaVisaRewards,
  ioniaGiftCards,
  kado,
  letsexchange,
  libertyx,
  lifi,
  maya,
  moonpay,
  paybis,
  paytrie,
  rango,
  safello,
  sideshift,
  simplex,
  swapuz,
  switchain,
  thorchain,
  transak,
  wyre,
  xanpool
]
const QUERY_FREQ_MS = 60 * 1000
const MAX_CONCURRENT_QUERIES = 3
const BULK_FETCH_SIZE = 500
const snooze: Function = async (ms: number) =>
  await new Promise((resolve: Function) => setTimeout(resolve, ms))

const dbProgress = nanoDb.db.use('reports_progresscache')
const dbApps = nanoDb.db.use('reports_apps')
const dbSettings: nano.DocumentScope<unknown> = nanoDb.db.use(
  'reports_settings'
)

export async function queryEngine(): Promise<void> {
  while (true) {
    datelog('Starting query loop...')
    let disablePartnerQuery: DisablePartnerQuery = {
      plugins: {},
      appPartners: {}
    }
    try {
      const disablePartnerQueryDoc = await dbSettings.get('disablePartnerQuery')
      if (disablePartnerQueryDoc != null) {
        disablePartnerQuery = asDisablePartnerQuery(disablePartnerQueryDoc)
      }
    } catch (e) {
      datelog('Error getting disablePartnerQuery', e)
    }
    // get the contents of all reports_apps docs
    const query = {
      selector: {
        appId: { $exists: true }
      },
      limit: 1000000
    }
    const rawApps = await dbApps.find(query)
    const apps = asApps(rawApps.docs)
    // loop over every app
    for (const app of apps) {
      const semaphore = new Semaphore(MAX_CONCURRENT_QUERIES)
      if (config.soloAppIds != null && !config.soloAppIds.includes(app.appId)) {
        continue
      }
      let partnerStatus: string[] = []
      const runPlugins: RunPluginParams[] = []
      let remainingPlugins: RunPluginParams[] = []
      // loop over every pluginId that app uses
      for (const partnerId in app.partnerIds) {
        const pluginId = app.partnerIds[partnerId].pluginId ?? partnerId
        if (config.soloPartnerIds?.includes(partnerId) !== true) {
          if (disablePartnerQuery.plugins[pluginId]) {
            continue
          }
          const appPartnerId = `${app.appId}_${partnerId}`
          if (disablePartnerQuery.appPartners[appPartnerId]) {
            continue
          }
          if (
            config.soloPartnerIds != null &&
            !config.soloPartnerIds.includes(partnerId)
          ) {
            continue
          }
        }
        const runPluginParams: RunPluginParams = { app, partnerId, pluginId }
        runPlugins.push(runPluginParams)
        remainingPlugins.push(runPluginParams)
      }
      const promises: Array<Promise<void>> = []
      for (const runPluginParams of runPlugins) {
        await semaphore.acquire()
        const promise = runPlugin(runPluginParams)
          .then(status => {
            partnerStatus = [...partnerStatus, status]
          })
          .finally(() => {
            semaphore.release()
            // remove the plugin from the remaining plugins
            remainingPlugins = remainingPlugins.filter(
              plugin => plugin !== runPluginParams
            )
            if (remainingPlugins.length > 0) {
              datelog(
                `REMAINING PLUGINS for ${app.appId}:`,
                remainingPlugins.map(plugin => plugin.partnerId).join(', ')
              )
            }
          })
        promises.push(promise)
      }
      await Promise.all(promises)
      datelog(partnerStatus.join('\n'))
    }
    datelog(`Snoozing for ${QUERY_FREQ_MS} milliseconds`)
    await snooze(QUERY_FREQ_MS)
  }
}

const checkUpdateTx = (oldTx: StandardTx, newTx: StandardTx): string[] => {
  const fields = [
    'status',
    'depositTxid',
    'depositChainPluginId',
    'depositEvmChainId',
    'depositTokenId',
    'payoutTxid',
    'payoutChainPluginId',
    'payoutEvmChainId',
    'payoutTokenId'
  ] as const
  const changedFields: string[] = []
  for (const field of fields) {
    if (oldTx[field] !== newTx[field]) changedFields.push(field)
  }
  return changedFields
}

const filterAddNewTxs = async (
  pluginId: string,
  dbTransactions: nano.DocumentScope<StandardTx>,
  docIds: string[],
  transactions: StandardTx[],
  log: ScopedLog
): Promise<void> => {
  if (docIds.length < 1 || transactions.length < 1) return
  const queryResults = await dbTransactions.fetch(
    { keys: docIds },
    { include_docs: true }
  )

  const newDocs: DbTx[] = []
  for (const docId of docIds) {
    const queryResult = queryResults.rows.find(
      doc => 'id' in doc && doc.id === docId && doc.doc != null
    )
    const orderId = docId.split(':')[1] ?? ''
    const tx = transactions.find(tx => tx.orderId === orderId)
    if (tx == null) {
      throw new Error(`Cant find tx from docId ${docId}`)
    }

    if (
      queryResult == null ||
      !('doc' in queryResult) ||
      queryResult.doc == null
    ) {
      // Get the full transaction
      const newObj = { _id: docId, _rev: undefined, ...tx }

      // replace all fields with non-standard names
      newObj.depositCurrency = standardizeNames(newObj.depositCurrency)
      newObj.payoutCurrency = standardizeNames(newObj.payoutCurrency)

      log(`[filterAddNewTxs] new doc id: ${newObj._id}`)
      newDocs.push(newObj)
    } else {
      const changedFields = checkUpdateTx(queryResult.doc, tx)
      if (changedFields.length > 0) {
        const oldStatus = queryResult.doc?.status
        const newStatus = tx.status
        const newObj = { _id: docId, _rev: queryResult.doc?._rev, ...tx }
        newDocs.push(newObj)
        log(
          `[filterAddNewTxs] updated doc id: ${
            newObj._id
          } ${oldStatus} -> ${newStatus} [${changedFields.join(', ')}]`
        )
      }
    }
  }

  try {
    await promiseTimeout(
      'pagination',
      pagination(newDocs, dbTransactions, log),
      log
    )
  } catch (e) {
    log.error('[filterAddNewTxs] Error doing bulk transaction insert', e)
    throw e
  }
}

async function insertTransactions(
  transactions: StandardTx[],
  pluginId: string,
  log: ScopedLog
): Promise<any> {
  const dbTransactions: nano.DocumentScope<StandardTx> = nanoDb.db.use(
    'reports_transactions'
  )
  let docIds: string[] = []
  let startIndex = 0
  for (let i = 0; i < transactions.length; i++) {
    const transaction = transactions[i]
    transaction.orderId = transaction.orderId.toLowerCase()
    const key = `${pluginId}:${transaction.orderId}`
    docIds.push(key)

    // Collect a batch of docIds
    if (docIds.length < BULK_FETCH_SIZE) continue

    log(`[insertTransactions] ${startIndex} to ${i} of ${transactions.length}`)
    await filterAddNewTxs(pluginId, dbTransactions, docIds, transactions, log)
    docIds = []
    startIndex = i + 1
  }
  await filterAddNewTxs(pluginId, dbTransactions, docIds, transactions, log)
}

interface RunPluginParams {
  app: ReturnType<typeof asApp>
  partnerId: string
  pluginId: string
}

async function runPlugin(params: RunPluginParams): Promise<string> {
  const { app, partnerId, pluginId } = params
  const start = Date.now()
  const log = createScopedLog(app.appId, partnerId)
  let errorText = ''
  try {
    // obtains function that corresponds to current pluginId
    const plugin = plugins.find(plugin => plugin.pluginId === pluginId)
    // if current plugin is not within the list of partners skip to next
    if (plugin === undefined) {
      errorText = `[runPlugin] ${partnerId} Missing or disabled plugin`
      log(errorText)
      return errorText
    }

    // get progress cache to see where previous query ended
    log(`[runPlugin] Starting with plugin:${pluginId}`)
    const progressCacheFileName = `${app.appId.toLowerCase()}:${partnerId}`
    const out = await dbProgress.get(progressCacheFileName).catch(e => {
      if (e.error != null && e.error === 'not_found') {
        log(`[runPlugin] Previous Progress Record Not Found`)
        return {}
      } else {
        log.error('[runPlugin] Error fetching progress', e)
      }
    })

    // initialize progress settings if unrecognized format
    let progressSettings: ReturnType<typeof asProgressSettings>
    try {
      progressSettings = asProgressSettings(out)
    } catch (e) {
      progressSettings = {
        progressCache: {},
        _id: undefined,
        _rev: undefined
      }
    }

    // set apiKeys and settings for use in partner's function
    const { apiKeys } = app.partnerIds[partnerId]
    const settings = progressSettings.progressCache
    log(`[runPlugin] Querying`)
    // run the plugin function
    const result = await promiseTimeout(
      'queryFunc',
      plugin.queryFunc({
        apiKeys,
        settings,
        log
      }),
      log
    )
    log(`[runPlugin] Successful query`)

    await promiseTimeout(
      'insertTransactions',
      insertTransactions(result.transactions, `${app.appId}_${partnerId}`, log),
      log
    ).catch(e => {
      throw new Error(`Error inserting transactions: ${String(e)}`)
    })
    progressSettings.progressCache = result.settings
    progressSettings._id = progressCacheFileName

    // Attempt to insert progress with retry on conflict
    const maxAttempts = 2
    let attempt = 0
    while (attempt < maxAttempts) {
      attempt++
      try {
        await promiseTimeout(
          `dbProgress.insert (attempt ${attempt})`,
          dbProgress.insert(progressSettings),
          log
        )
        break
      } catch (e) {
        const err: any = e
        const isConflict = err.statusCode === 409 || err.error === 'conflict'
        if (isConflict && attempt < maxAttempts) {
          log(`[runPlugin] Document conflict detected, re-reading and retrying`)
          const updatedDoc = await dbProgress.get(progressCacheFileName)
          progressSettings._rev = updatedDoc._rev
          continue
        }
        throw new Error(`Error inserting progress: ${String(e)}`)
      }
    }
    // Returning a successful completion message
    const completionTime = (Date.now() - start) / 1000
    const successfulCompletionMessage = `[runPlugin] ${partnerId} Successful update in ${completionTime} seconds.`
    log(successfulCompletionMessage)
    return successfulCompletionMessage
  } catch (e) {
    errorText = `[runPlugin] ${partnerId} Error: ${String(e)}`
    log.error(errorText)
    return errorText
  }
}