redis integration for notifications

2024-06-29 21:46:13 +05:00
parent f661499e77
commit 41d2ab7bba
7 changed files with 119 additions and 5 deletions
@@ -54,6 +54,7 @@
    "eosjs": "^22.1.0",
    "express": "^4.19.2",
    "express-async-errors": "^3.1.1",
+    "ioredis": "^5.4.1",
    "mongodb": "^6.5.0",
    "node-fetch": "^3.3.2",
    "typedoc": "^0.25.13",
@@ -29,6 +29,9 @@ importers:
      express-async-errors:
        specifier: ^3.1.1
        version: 3.1.1(express@4.19.2)
+      ioredis:
+        specifier: ^5.4.1
+        version: 5.4.1
      mongodb:
        specifier: ^6.5.0
        version: 6.5.0
@@ -570,6 +573,9 @@ packages:
  '@humanwhocodes/object-schema@2.0.3':
    resolution: {integrity: sha512-93zYdMES/c1D69yZiKDBj0V24vqNzB/koF26KPaagAfd3P/4gUlh3Dys5ogAK+Exi9QyzlD8x/08Zt7wIKcDcA==}

+  '@ioredis/commands@1.2.0':
+    resolution: {integrity: sha512-Sx1pU8EM64o2BrqNpEO1CNLtKQwyhuXuqyfH7oGKCk+1a33d2r5saW8zNwm3j6BTExtjrv2BxTgzzkMwts6vGg==}
+
  '@isaacs/cliui@8.0.2':
    resolution: {integrity: sha512-O8jcjabXaleOG9DQ0+ARXWZBTfnP4WNAqzuiJK7ll44AmxGKv/J2M4TPjxjY3znBCfvBXFzucm1twdyFybFqEA==}
    engines: {node: '>=12'}
@@ -1204,6 +1210,10 @@ packages:
    resolution: {integrity: sha512-BSeNnyus75C4//NQ9gQt1/csTXyo/8Sb+afLAkzAptFuMsod9HFokGNudZpi/oQV73hnVK+sR+5PVRMd+Dr7YQ==}
    engines: {node: '>=12'}

+  cluster-key-slot@1.1.2:
+    resolution: {integrity: sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==}
+    engines: {node: '>=0.10.0'}
+
  color-convert@1.9.3:
    resolution: {integrity: sha512-QfAUtd+vFdAtFQcC8CCyYt1fYWxSqAiK2cSD6zDB8N3cpsEBAvRxp9zOGg6G/SHHJYAT88/az/IuDGALsNVbGg==}

@@ -1385,6 +1395,10 @@ packages:
    resolution: {integrity: sha512-ZySD7Nf91aLB0RxL4KGrKHBXl7Eds1DAmEdcoVawXnLD7SDhpNgtuII2aAkg7a7QS41jxPSZ17p4VdGnMHk3MQ==}
    engines: {node: '>=0.4.0'}

+  denque@2.1.0:
+    resolution: {integrity: sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==}
+    engines: {node: '>=0.10'}
+
  depd@2.0.0:
    resolution: {integrity: sha512-g7nH6P6dyDioJogAAGprGpCtVImJhpPk/roCzdb3fIh61/s/nPsfR6onyMwkCAR/OlC3yBC0lESvUoQEAssIrw==}
    engines: {node: '>= 0.8'}
@@ -1973,6 +1987,10 @@ packages:
  inherits@2.0.4:
    resolution: {integrity: sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ==}

+  ioredis@5.4.1:
+    resolution: {integrity: sha512-2YZsvl7jopIa1gaePkeMtd9rAcSjOOjPtpcLlOeusyO+XH2SK5ZcT+UCrElPP+WVIInh2TzeI4XW9ENaSLVVHA==}
+    engines: {node: '>=12.22.0'}
+
  ipaddr.js@1.9.1:
    resolution: {integrity: sha512-0KI/607xoxSToH7GjN1FfSbLoU0+btTicjsQSWQlh/hZykN8KpmMf7uYwPW3R+akZ6R/w18ZlXSHBYXiYUPO3g==}
    engines: {node: '>= 0.10'}
@@ -2164,6 +2182,12 @@ packages:
    resolution: {integrity: sha512-gvVijfZvn7R+2qyPX8mAuKcFGDf6Nc61GdvGafQsHL0sBIxfKzA+usWn4GFC/bk+QdwPUD4kWFJLhElipq+0VA==}
    engines: {node: ^12.20.0 || ^14.13.1 || >=16.0.0}

+  lodash.defaults@4.2.0:
+    resolution: {integrity: sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==}
+
+  lodash.isarguments@3.1.0:
+    resolution: {integrity: sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg==}
+
  lodash.memoize@4.1.2:
    resolution: {integrity: sha512-t7j+NzmgnQzTAYXcsHYLgimltOV1MXHtlOWf6GjL9Kj8GK5FInw5JotxvbOs+IvV1/Dzo04/fCGfLVs7aXb4Ag==}

@@ -2853,6 +2877,14 @@ packages:
    resolution: {integrity: sha512-hOS089on8RduqdbhvQ5Z37A0ESjsqz6qnRcffsMU3495FuTdqSm+7bhJ29JvIOsBDEEnan5DPu9t3To9VRlMzA==}
    engines: {node: '>=8.10.0'}

+  redis-errors@1.2.0:
+    resolution: {integrity: sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==}
+    engines: {node: '>=4'}
+
+  redis-parser@3.0.0:
+    resolution: {integrity: sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==}
+    engines: {node: '>=4'}
+
  regexp-tree@0.1.27:
    resolution: {integrity: sha512-iETxpjK6YoRWJG5o6hXLwvjYAoW+FEZn9os0PD/b6AP6xQwsa/Y7lCVgIixBbUPMfhu+i2LtdeAqVTgGlQarfA==}
    hasBin: true
@@ -3044,6 +3076,9 @@ packages:
  stackback@0.0.2:
    resolution: {integrity: sha512-1XMJE5fQo1jGH6Y/7ebnwPOBEkIEnT4QF32d5R1+VXdXveM0IBMJt8zfaxX1P3QhVwrYe+576+jkANtSS2mBbw==}

+  standard-as-callback@2.1.0:
+    resolution: {integrity: sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==}
+
  statuses@2.0.1:
    resolution: {integrity: sha512-RwNA9Z/7PrK06rYLIzFMlaF+l73iwpzsqRIFgbMLbTcLD6cOao82TaWefPXQvB2fOC4AjuYSEndS7N/mTCbkdQ==}
    engines: {node: '>= 0.8'}
@@ -3878,6 +3913,8 @@ snapshots:

  '@humanwhocodes/object-schema@2.0.3': {}

+  '@ioredis/commands@1.2.0': {}
+
  '@isaacs/cliui@8.0.2':
    dependencies:
      string-width: 5.1.2
@@ -4600,6 +4637,8 @@ snapshots:
      strip-ansi: 6.0.1
      wrap-ansi: 7.0.0

+  cluster-key-slot@1.1.2: {}
+
  color-convert@1.9.3:
    dependencies:
      color-name: 1.1.3
@@ -4777,6 +4816,8 @@ snapshots:

  delayed-stream@1.0.0: {}

+  denque@2.1.0: {}
+
  depd@2.0.0: {}

  destr@2.0.3: {}
@@ -5547,6 +5588,20 @@ snapshots:

  inherits@2.0.4: {}

+  ioredis@5.4.1:
+    dependencies:
+      '@ioredis/commands': 1.2.0
+      cluster-key-slot: 1.1.2
+      debug: 4.3.4
+      denque: 2.1.0
+      lodash.defaults: 4.2.0
+      lodash.isarguments: 3.1.0
+      redis-errors: 1.2.0
+      redis-parser: 3.0.0
+      standard-as-callback: 2.1.0
+    transitivePeerDependencies:
+      - supports-color
+
  ipaddr.js@1.9.1: {}

  is-alphabetical@1.0.4: {}
@@ -5715,6 +5770,10 @@ snapshots:
    dependencies:
      p-locate: 6.0.0

+  lodash.defaults@4.2.0: {}
+
+  lodash.isarguments@3.1.0: {}
+
  lodash.memoize@4.1.2: {}

  lodash.merge@4.6.2: {}
@@ -6331,6 +6390,12 @@ snapshots:
    dependencies:
      picomatch: 2.3.1

+  redis-errors@1.2.0: {}
+
+  redis-parser@3.0.0:
+    dependencies:
+      redis-errors: 1.2.0
+
  regexp-tree@0.1.27: {}

  regjsparser@0.10.0:
@@ -6541,6 +6606,8 @@ snapshots:

  stackback@0.0.2: {}

+  standard-as-callback@2.1.0: {}
+
  statuses@2.0.1: {}

  std-env@3.7.0: {}
@@ -1,5 +1,7 @@
 import type { Database } from '../../Database'
+import { publishEvent } from '../../RedisNotifier'
 import type { EosioShipReaderResolved, IAction } from '../../Types'
+import { subsribedActions } from '../../config'
 import { ActionParserFactory } from '../Factory/ActionFactory'

 export async function ActionsParser(db: Database, reader: EosioShipReaderResolved) {
@@ -9,8 +11,14 @@ export async function ActionsParser(db: Database, reader: EosioShipReaderResolve
    console.log(`\nACTION - account: ${action.account}, name: ${action.name}, authorization: ${JSON.stringify(action.authorization)}, data: ${JSON.stringify(action.data)}`)

    const parser = ActionParserFactory.create(action.account, action.name)
-    if (parser)
+    const source = subsribedActions.find(el => el.action === action.name && el.code === action.account)
+
+    if (parser) {
      await parser.process(db, action)
+
+      if (source?.notify)
+        await publishEvent('action', action)
+    }
  })

  console.log('Подписка на действия активирована')
@@ -1,5 +1,7 @@
 import type { Database } from '../../Database'
+import { publishEvent } from '../../RedisNotifier'
 import type { EosioShipReaderResolved, IDelta } from '../../Types'
+import { subsribedTables } from '../../config'
 import { DeltaParserFactory } from '../Factory'

 export async function DeltasParser(db: Database, reader: EosioShipReaderResolved) {
@@ -7,10 +9,15 @@ export async function DeltasParser(db: Database, reader: EosioShipReaderResolved

  rows$.subscribe(async (delta: IDelta) => {
    console.log(`\nDELTA - code: ${delta.code}, scope: ${delta.scope}, table: ${delta.table}, primary_key: ${delta.primary_key}, data: ${JSON.stringify(delta.value)}`)
+    const source = subsribedTables.find(el => el.code === delta.code && el.table === delta.table)

    const parser = DeltaParserFactory.create(delta.code, delta.scope, delta.table)
-    if (parser)
+    if (parser) {
      await parser.process(db, delta)
+
+      if (source?.notify)
+        await publishEvent('delta', delta)
+    }
  })

  console.log('Подписка на дельты активирована')
@@ -0,0 +1,13 @@
+import Redis from 'ioredis'
+import { redisPort, redisStreamLimit } from '../config'
+
+const redis = new Redis(redisPort)
+
+const streamName = 'notifications'
+
+export async function publishEvent(type: string, event: object) {
+  const message = JSON.stringify({ type, event })
+
+  await redis.xadd(streamName, '*', 'event', message)
+  await redis.xtrim(streamName, 'MAXLEN', '~', redisStreamLimit)
+}
@@ -8,6 +8,19 @@ export { EosioReaderAction as IAction } from '@blockmatic/eosio-ship-reader'
 export type EosioShipReader = ReturnType<typeof createEosioShipReader>
 export type EosioShipReaderResolved = Awaited<EosioShipReader>

+export interface IDeltaConfig {
+  code: string
+  table: string
+  scope?: string
+  notify?: boolean
+}
+
+export interface IActionConfig {
+  code: string
+  action: string
+  notify?: boolean
+}
+
 export interface IParserDelta {
  process: (db: Database, delta: IDelta) => Promise<void>
 }
@@ -1,4 +1,5 @@
 import dotenv from 'dotenv'
+import type { IActionConfig, IDeltaConfig } from './Types'

 dotenv.config()

@@ -15,9 +16,10 @@ export const shipApi = getEnvVar('SHIP')
 export const mongoUri = getEnvVar('MONGO_EXPLORER_URI')
 export const startBlock = getEnvVar('START_BLOCK')
 export const finishBlock = getEnvVar('FINISH_BLOCK')
-
+export const redisPort = getEnvVar('REDIS_PORT')
+export const redisStreamLimit = Number(getEnvVar('REDIS_STREAM_LIMIT'))
 // --------------------------
-export const subsribedTables = [
+export const subsribedTables: IDeltaConfig[] = [
  // документы
  { code: 'draft', table: 'drafts' },
  { code: 'draft', table: 'translations' },
@@ -35,7 +37,10 @@ export const subsribedTables = [
  { code: 'registrator', table: 'orgs' },
 ]

-export const subsribedActions = [
+export const subsribedActions: IActionConfig[] = [
+  { code: 'eosio.token', action: 'transfer', notify: true },
+  { code: 'registrator', action: 'confirmreg', notify: true },
+
  { code: 'soviet', action: 'votefor' },
  { code: 'soviet', action: 'voteagainst' },
  { code: 'soviet', action: 'newsubmitted' },