Initial version

eslint.config.js

@@ -21,6 +21,7 @@ export default tseslint.config({
     '@typescript-eslint/no-unsafe-assignment': 'error',
     '@typescript-eslint/no-unsafe-call': 'error',
     '@typescript-eslint/no-unsafe-member-access': 'error',
-    '@typescript-eslint/no-unsafe-return': 'error'
+    '@typescript-eslint/no-unsafe-return': 'error',
+    '@typescript-eslint/no-explicit-any': 'off'
   }
 })

src/events/channels/transactionsChannel.ts

@@ -0,0 +1,167 @@
+import EventEmitter from 'events'
+import { AnyTransaction, WebSocketClient } from 'adamant-api'
+import { schedule } from 'node-cron'
+import { logger } from '../../modules/logger.js'
+import { config } from '../../config/index.js'
+import { adamantClient } from '../../modules/adamantClient.js'
+import { prisma } from '../../modules/prisma.js'
+import { JobName } from '@prisma/client'
+import { isSignalTx } from '../../services/parser/isSignalTx.js'
+import { isTxToNotify } from '../../services/parser/isTxToNotify.js'
+import { ChatMessageTransaction } from 'adamant-api/dist/api/generated.js'
+
+class TransactionsChannel extends EventEmitter {
+  // Specify types of events
+  override emit(event: 'newSignalMessage', tx: ChatMessageTransaction): boolean
+  override emit(event: 'newMessage', tx: AnyTransaction): boolean
+  override emit(event: string, ...args: never[]): boolean {
+    return super.emit(event, ...args)
+  }
+
+  override on(
+    event: 'newSignalMessage',
+    listener: (tx: ChatMessageTransaction) => void
+  ): this
+  override on(event: 'newMessage', listener: (tx: AnyTransaction) => void): this
+  override on(event: string, listener: (...args: any[]) => void): this {
+    return super.on(event, listener)
+  }
+
+  private isLocked: boolean
+  private processedTxs: { [key: string]: AnyTransaction } = {} // cache for processed transactions
+  adamantSocket?: WebSocketClient
+
+  constructor() {
+    super()
+    this.isLocked = false
+  }
+
+  initSocket() {
+    adamantClient.initSocket({
+      wsType: 'ws',
+      admAddress: config.adamantAccount.address
+    })
+    logger.info(
+      `Adamant Client socket initialized on ${config.adamantAccount.address} address`
+    )
+
+    if (adamantClient.socket) {
+      adamantClient.socket.on((tx) => {
+        this.handleTransaction(tx)
+      })
+      adamantClient.socket.catch((error) => logger.error(error))
+
+      this.adamantSocket = adamantClient.socket
+    }
+  }
+
+  startJob() {
+    logger.info(
+      `Spawned transaction parser job with ${config.app.txCheckInterval} interval`
+    )
+    return schedule(config.app.txCheckInterval, async () => {
+      if (this.isLocked) return
+
+      this.isLocked = true
+
+      try {
+        const getHeightResponse = await adamantClient.getHeight()
+        const currentHeight = getHeightResponse.success
+          ? getHeightResponse.height
+          : 0
+
+        let jobStatus = await prisma.cronJobStatus.findFirst({
+          where: { jobName: JobName.TRANSACTIONS }
+        })
+
+        if (!jobStatus) {
+          jobStatus = await prisma.cronJobStatus.create({
+            data: {
+              jobName: JobName.TRANSACTIONS,
+              state: JSON.stringify({
+                lastHeight: currentHeight
+              })
+            }
+          })
+        }
+
+        const lastCheckHeight = (
+          JSON.parse(jobStatus.state as string) as { lastHeight: number }
+        ).lastHeight
+
+        // Determine fetch interval
+        let heightToFetch = lastCheckHeight + config.app.heightSkipPerHeight
+
+        if (heightToFetch > currentHeight) {
+          this.isLocked = false
+          return
+        }
+
+        if (currentHeight - heightToFetch > 1) {
+          // If the gap is more than 1 block, fetch transactions in chunks of 1000 blocks
+          heightToFetch = lastCheckHeight + 1000
+
+          if (currentHeight - heightToFetch < 0) {
+            // if the gap reaches currenHeight and more, than set it to currentHeight
+            heightToFetch = currentHeight
+          }
+        }
+
+        const txs = await adamantClient.getTransactions({
+          fromHeight: lastCheckHeight,
+          and: {
+            toHeight: heightToFetch
+          },
+          returnAsset: 1
+        })
+
+        if (!txs.success) {
+          this.isLocked = false
+          return
+        }
+
+        txs.transactions.forEach((tx) => {
+          if (tx.height < currentHeight - config.notify.latestHeightToNotify) {
+            // skip if transaction is too old
+            return
+          }
+
+          this.handleTransaction(tx)
+        })
+
+        await prisma.cronJobStatus.update({
+          where: { jobName: JobName.TRANSACTIONS },
+          data: {
+            state: JSON.stringify({
+              lastHeight: heightToFetch
+            })
+          }
+        })
+      } catch (error) {
+        logger.error(error, 'Error while running transactions channel job')
+      } finally {
+        this.isLocked = false
+      }
+    })
+  }
+
+  handleTransaction(tx: AnyTransaction) {
+    if (this.processedTxs[tx.id]) {
+      delete this.processedTxs[tx.id] // removing from cache because we got tx again from rest api or socket
+      return
+    }
+
+    this.processedTxs[tx.id] = tx
+
+    if (isSignalTx(tx)) {
+      logger.info(
+        `Got signal transaction to (un)subscribe to notifications, txId: ${tx.id}, processing...`
+      )
+      this.emit('newSignalMessage', tx as ChatMessageTransaction)
+    } else if (isTxToNotify(tx)) {
+      this.emit('newMessage', tx)
+    }
+  }
+}
+
+export const transactionsChannel = new TransactionsChannel()

src/events/handlers.ts

@@ -1,37 +1,17 @@
-import {
-  notifyMessagesChannel,
-  signalMessagesChannel,
-  transactionsChannel
-} from './channels.js'
-import { txsParser } from '../services/txsParser.js'
-import { AnyTransaction } from 'adamant-api'
+import { transactionsChannel } from './channels/transactionsChannel.js'
 import { logger } from '../modules/logger.js'
-import { ChatMessageTransaction } from 'adamant-api/dist/api/generated.js'
 import { processSignalTransaction } from '../services/processSignalTransaction.js'
 import { processTransactionToNotify } from '../services/processTransactionToNotify.js'
 import { prisma } from '../modules/prisma.js'
 
 export const spawnEventHandlers = () => {
-  const processedTxs: { [key: string]: AnyTransaction } = {} // cache for processed transactions
-
-  transactionsChannel.on('newTransaction', (tx: AnyTransaction) => {
-    if (processedTxs[tx.id]) {
-      delete processedTxs[tx.id] // removing from cache because we got tx again from rest api or socket
-      return
-    }
-
-    processedTxs[tx.id] = tx
-
-    txsParser(tx).catch((e) => logger.error(e, 'Failed to parsed transaction'))
-  })
-
-  signalMessagesChannel.on('newSignalMessage', (tx: ChatMessageTransaction) => {
+  transactionsChannel.on('newSignalMessage', (tx) => {
     processSignalTransaction(tx).catch((e) =>
       logger.error(e, 'Failed to process signal transaction')
     )
   })
 
-  notifyMessagesChannel.on('newMessage', async (tx: AnyTransaction) => {
+  transactionsChannel.on('newMessage', async (tx) => {
     let devices = await prisma.device.findMany({
       where: { admAddress: tx.recipientId }
     })

src/jobs/index.ts

@@ -1,8 +1,8 @@
-import { spawnTransactionsJobs } from './transactionsJobs.js'
+import { transactionsChannel } from '../events/channels/transactionsChannel.js'
 import { spawnRetryNotifyJob } from './retryNotifyJob.js'
 
 export const spawnJobs = () => {
-  const transactionJob = spawnTransactionsJobs()
+  const transactionJob = transactionsChannel.startJob()
   const retryNotifyJob = spawnRetryNotifyJob()
 
   transactionJob.start()

src/jobs/retryNotifyJob.ts

@@ -17,30 +17,34 @@ export const spawnRetryNotifyJob = () => {
 
     isLocked = true
 
-    const transactionsToNotify = await prisma.notifyTransaction.findMany({
-      where: {
-        isNotified: false,
-        lastNotifyDate: { not: null }
-      },
-      include: {
-        device: true
-      }
-    })
-
-    await Promise.all(
-      transactionsToNotify.map(async (transaction) => {
-        const tx = JSON.parse(transaction.admTx as string) as AnyTransaction
-
-        logger.info(
-          `Got notification that failed to send, processing... DeviceId: ${transaction.device.id}, provider: ${transaction.device.pushServiceProvider}, admTxId: ${tx.id}`
-        )
-
-        processTransactionToNotify(tx, transaction.device).catch((e) =>
-          logger.error(e, 'Failed to process failed transaction to notify')
-        )
+    try {
+      const transactionsToNotify = await prisma.notifyTransaction.findMany({
+        where: {
+          isNotified: false,
+          lastNotifyDate: { not: null }
+        },
+        include: {
+          device: true
+        }
       })
-    )
 
-    isLocked = false
+      await Promise.all(
+        transactionsToNotify.map(async (transaction) => {
+          const tx = JSON.parse(transaction.admTx as string) as AnyTransaction
+
+          logger.info(
+            `Got notification that failed to send, processing... DeviceId: ${transaction.device.id}, provider: ${transaction.device.pushServiceProvider}, admTxId: ${tx.id}`
+          )
+
+          processTransactionToNotify(tx, transaction.device).catch((e) =>
+            logger.error(e, 'Failed to process failed transaction to notify')
+          )
+        })
+      )
+    } catch (err) {
+      logger.error(err, 'Error while running retry notify job')
+    } finally {
+      isLocked = false
+    }
   })
 }