fix: avoiding IDs override/conflict while creating concurrent runtime notifications (#342)

* fix: avoiding IDs override while creating conturrent runtime notifications

* create record in 'getNextIdForEntity' function in overlay for entityname, if it does not exist

* fix: notifications unit tests

Author: zeeshanakram3

src/mappings/token/utils.ts

@@ -327,8 +327,8 @@ export async function getHolderAccountsForToken(
   const holders = await em.getRepository(TokenAccount).findBy({ tokenId })
 
   const holdersMemberIds = holders
-    .filter((follower) => follower?.memberId)
-    .map((follower) => follower.memberId as string)
+    .filter((holder) => holder?.memberId)
+    .map((holder) => holder.memberId as string)
 
   const limit = pLimit(10) // Limit to 10 concurrent promises
   const holdersAccounts: (Account | null)[] = await Promise.all(
@@ -347,17 +347,17 @@ export async function notifyTokenHolders(
   event?: Event,
   dispatchBlock?: number
 ) {
-  const holdersAccounts = await getHolderAccountsForToken(em, tokenId)
+  const holderAccounts = await getHolderAccountsForToken(em, tokenId)
 
   const limit = pLimit(10) // Limit to 10 concurrent promises
 
   await Promise.all(
-    holdersAccounts.map((holdersAccount) =>
+    holderAccounts.map((holderAccount) =>
       limit(() =>
         addNotification(
           em,
-          holdersAccount,
-          new MemberRecipient({ membership: holdersAccount.membershipId }),
+          holderAccount,
+          new MemberRecipient({ membership: holderAccount.membershipId }),
           notificationType,
           event,
           dispatchBlock

src/tests/integration/notifications.test.ts

@@ -1,12 +1,14 @@
-import { EntityManager } from 'typeorm'
-import { IMemberRemarked, ReactVideo, MemberRemarked } from '@joystream/metadata-protobuf'
+import { IMemberRemarked, MemberRemarked, ReactVideo } from '@joystream/metadata-protobuf'
 import { AnyMetadataClass } from '@joystream/metadata-protobuf/types'
-import { defaultTestBlock, populateDbWithSeedData } from './testUtils'
-import { globalEm } from '../../utils/globalEm'
-import {
-  excludeChannelService,
-  verifyChannelService,
-} from '../../server-extension/resolvers/ChannelsResolver'
+import { Store } from '@subsquid/typeorm-store'
+import { expect } from 'chai'
+import { config as dontenvConfig } from 'dotenv'
+import Long from 'long'
+import path from 'path'
+import { EntityManager } from 'typeorm'
+import { auctionBidMadeInner } from '../../mappings/content/nft'
+import { processMemberRemarkedEvent } from '../../mappings/membership'
+import { backwardCompatibleMetaID } from '../../mappings/utils'
 import {
   Account,
   Channel,
@@ -26,21 +28,19 @@ import {
   Video,
   VideoLiked,
 } from '../../model'
-import { expect } from 'chai'
+import { setFeaturedNftsInner } from '../../server-extension/resolvers/AdminResolver'
+import {
+  excludeChannelService,
+  verifyChannelService,
+} from '../../server-extension/resolvers/ChannelsResolver'
+import { excludeVideoService } from '../../server-extension/resolvers/VideosResolver'
+import { globalEm } from '../../utils/globalEm'
 import {
   OFFCHAIN_NOTIFICATION_ID_TAG,
   RUNTIME_NOTIFICATION_ID_TAG,
 } from '../../utils/notification/helpers'
-import { setFeaturedNftsInner } from '../../server-extension/resolvers/AdminResolver'
-import { auctionBidMadeInner } from '../../mappings/content/nft'
-import { EntityManagerOverlay } from '../../utils/overlay'
-import { Store } from '@subsquid/typeorm-store'
-import { processMemberRemarkedEvent } from '../../mappings/membership'
-import Long from 'long'
-import { backwardCompatibleMetaID } from '../../mappings/utils'
-import { config as dontenvConfig } from 'dotenv'
-import path from 'path'
-import { excludeVideoService } from '../../server-extension/resolvers/VideosResolver'
+import { AnyEntity, Constructor, EntityManagerOverlay } from '../../utils/overlay'
+import { defaultTestBlock, populateDbWithSeedData } from './testUtils'
 
 dontenvConfig({
   path: path.resolve(__dirname, './.env'),
@@ -50,9 +50,19 @@ const metadataToBytes = <T>(metaClass: AnyMetadataClass<T>, obj: T): Uint8Array
   return Buffer.from(metaClass.encode(obj).finish())
 }
 
-const getNextNotificationId = async (em: EntityManager, onchain: boolean) => {
+const getNextNotificationId = async (
+  store: EntityManager | EntityManagerOverlay,
+  onchain: boolean
+) => {
   const tag = onchain ? RUNTIME_NOTIFICATION_ID_TAG : OFFCHAIN_NOTIFICATION_ID_TAG
-  const row = await em.getRepository(NextEntityId).findOneBy({ entityName: tag })
+  if (store instanceof EntityManagerOverlay) {
+    const row = await store
+      .getRepository(NextEntityId as Constructor<NextEntityId & AnyEntity>)
+      .getOneBy({ entityName: tag })
+    const id = parseInt(row?.nextId.toString() || '1')
+    return id
+  }
+  const row = await store.getRepository(NextEntityId).findOneBy({ entityName: tag })
   const id = parseInt(row?.nextId.toString() || '1')
   return id
 }
@@ -69,6 +79,7 @@ describe('notifications tests', () => {
   let em: EntityManager
   before(async () => {
     em = await globalEm
+    overlay = await createOverlay()
     await populateDbWithSeedData()
   })
   describe('👉 YPP Verify channel', () => {
@@ -281,10 +292,9 @@ describe('notifications tests', () => {
 
     before(async () => {
       const bidAmount = BigInt(100000)
-      nextNotificationIdPre = await getNextNotificationId(em, true)
+      nextNotificationIdPre = await getNextNotificationId(overlay, true)
       notificationId = RUNTIME_NOTIFICATION_ID_TAG + '-' + nextNotificationIdPre
       nft = await em.getRepository(OwnedNft).findOneByOrFail({ videoId })
-      overlay = await createOverlay()
 
       await auctionBidMadeInner(
         overlay,
@@ -368,8 +378,7 @@ describe('notifications tests', () => {
       asV2001: ['3', metadataToBytes(MemberRemarked, metadataMessage), undefined],
     } as any
     before(async () => {
-      overlay = await createOverlay()
-      nextNotificationIdPre = await getNextNotificationId(em, true)
+      nextNotificationIdPre = await getNextNotificationId(overlay, true)
       notificationId = RUNTIME_NOTIFICATION_ID_TAG + '-' + nextNotificationIdPre.toString()
     })
     it('should process video liked and deposit notification', async () => {
@@ -381,7 +390,7 @@ describe('notifications tests', () => {
         event,
       })
 
-      const nextNotificationId = await getNextNotificationId(em, true)
+      const nextNotificationId = await getNextNotificationId(overlay, true)
       notification = (await overlay
         .getRepository(Notification)
         .getByIdOrFail(notificationId)) as Notification
@@ -422,8 +431,7 @@ describe('notifications tests', () => {
       asV2001: ['2', metadataToBytes(MemberRemarked, metadataMessage), undefined], // avoid comment author == creator
     } as any
     before(async () => {
-      overlay = await createOverlay()
-      nextNotificationIdPre = await getNextNotificationId(em, true)
+      nextNotificationIdPre = await getNextNotificationId(overlay, true)
       notificationId = RUNTIME_NOTIFICATION_ID_TAG + '-' + nextNotificationIdPre.toString()
     })
     it('should process comment to video and deposit notification', async () => {
@@ -435,7 +443,7 @@ describe('notifications tests', () => {
         event,
       })
 
-      const nextNotificationId = await getNextNotificationId(em, true)
+      const nextNotificationId = await getNextNotificationId(overlay, true)
       notification = (await overlay
         .getRepository(Notification)
         .getByIdOrFail(notificationId)) as Notification | null
@@ -485,7 +493,7 @@ describe('notifications tests', () => {
       } as any
 
       before(async () => {
-        nextNotificationIdPre = await getNextNotificationId(em, true)
+        nextNotificationIdPre = await getNextNotificationId(overlay, true)
         notificationId = RUNTIME_NOTIFICATION_ID_TAG + '-' + nextNotificationIdPre.toString()
 
         await processMemberRemarkedEvent({
@@ -500,7 +508,7 @@ describe('notifications tests', () => {
       describe('should process reply to comment and deposit notification', () => {
         let nextNotificationId: number
         before(async () => {
-          nextNotificationId = await getNextNotificationId(em, true)
+          nextNotificationId = await getNextNotificationId(overlay, true)
           notification = (await overlay
             .getRepository(Notification)
             .getByIdOrFail(notificationId)) as Notification | null

src/utils/nextEntityId.ts

@@ -1,15 +1,40 @@
 import { EntityManager } from 'typeorm'
 import { NextEntityId } from '../model'
+import { AnyEntity, Constructor, EntityManagerOverlay } from './overlay'
 
-// used to retrieve the next id for an entity
-export async function getNextIdForEntity(em: EntityManager, entityName: string): Promise<number> {
+// used to retrieve the next id for an entity from NextEntityId table using either EntityManager or Overlay
+export async function getNextIdForEntity(
+  store: EntityManager | EntityManagerOverlay,
+  entityName: string
+): Promise<number> {
+  // Get next entity id from overlay (this will mostly be used in the mappings context)
+  if (store instanceof EntityManagerOverlay) {
+    const row = await store
+      .getRepository(NextEntityId as Constructor<NextEntityId & AnyEntity>)
+      .getOneBy({ entityName: entityName })
+
+    const id = parseInt(row?.nextId.toString() || '1')
+
+    // Update the id to be the next one in the overlay
+    if (row) {
+      row.nextId++
+    } else {
+      store
+        .getRepository(NextEntityId as Constructor<NextEntityId & AnyEntity>)
+        .new({ entityName, nextId: id + 1 })
+    }
+
+    return id
+  }
+
+  // Get next entity id from EntityManager (this will mostly be used in the graphql-server/auth-api context)
   let row: NextEntityId | null
   if (process.env.TESTING === 'true' || process.env.TESTING === '1') {
-    row = await em.getRepository(NextEntityId).findOne({
+    row = await store.getRepository(NextEntityId).findOne({
       where: { entityName },
     })
   } else {
-    row = await em.getRepository(NextEntityId).findOne({
+    row = await store.getRepository(NextEntityId).findOne({
       where: { entityName },
       lock: { mode: 'pessimistic_write' },
     })

src/utils/notification/helpers.ts

@@ -159,7 +159,7 @@ async function addOffChainNotification(
   const nextNotificationId = await getNextIdForEntity(em, OFFCHAIN_NOTIFICATION_ID_TAG)
 
   const notification = createNotification(
-    OFFCHAIN_NOTIFICATION_ID_TAG + '-' + nextNotificationId.toString(),
+    `${OFFCHAIN_NOTIFICATION_ID_TAG}-${nextNotificationId}`,
     account.id,
     recipient,
     notificationType,
@@ -186,21 +186,22 @@ async function addRuntimeNotification(
   event: Event,
   dispatchBlock?: number
 ) {
-  const em = overlay.getEm()
   // get notification Id from orion_db in any case
-  const nextNotificationId = await getNextIdForEntity(em, RUNTIME_NOTIFICATION_ID_TAG)
+  const nextNotificationId = await getNextIdForEntity(overlay, RUNTIME_NOTIFICATION_ID_TAG)
+
+  const runtimeNotificationId = `${RUNTIME_NOTIFICATION_ID_TAG}-${nextNotificationId}`
 
   // check that on-notification is not already present in orion_db in case the processor has been restarted (but not orion_db)
   const existingNotification = await overlay
     .getRepository(Notification)
-    .getById(nextNotificationId.toString())
+    .getById(runtimeNotificationId)
 
   if (existingNotification) {
     return
   }
 
   const notification = createNotification(
-    RUNTIME_NOTIFICATION_ID_TAG + '-' + nextNotificationId.toString(),
+    runtimeNotificationId,
     account.id,
     recipient,
     notificationType,
@@ -216,8 +217,6 @@ async function addRuntimeNotification(
     await createEmailNotification(overlay, notification)
   }
 
-  await saveNextNotificationId(em, nextNotificationId + 1, RUNTIME_NOTIFICATION_ID_TAG)
-
   return notification.id
 }
 

src/utils/notification/index.ts

@@ -1,7 +1,5 @@
 export {
+  addNotification,
   defaultNotificationPreferences,
   preferencesForNotification,
-  addNotification,
 } from './helpers'
-// export * from './notificationTexts'
-// export * from './notificationLinks'

src/utils/overlay.ts

@@ -345,16 +345,16 @@ export class EntityManagerOverlay {
   constructor(
     private em: EntityManager,
     private nextEntityIds: NextEntityId[],
-    private afterDbUpdte: (em: EntityManager) => Promise<void>
+    private afterDbUpdate: (em: EntityManager) => Promise<void>
   ) {}
 
-  public static async create(store: Store, afterDbUpdte: (em: EntityManager) => Promise<void>) {
+  public static async create(store: Store, afterDbUpdate: (em: EntityManager) => Promise<void>) {
     // FIXME: This is a little hacky, but we really need to access the underlying EntityManager
     const em = await (store as unknown as { em: () => Promise<EntityManager> }).em()
     // Add "admin" schema to search path in order to be able to access "hidden" entities
     await em.query('SET search_path TO admin,public')
     const nextEntityIds = await em.find(NextEntityId, {})
-    return new EntityManagerOverlay(em, nextEntityIds, afterDbUpdte)
+    return new EntityManagerOverlay(em, nextEntityIds, afterDbUpdate)
   }
 
   public totalCacheSize() {
@@ -400,6 +400,6 @@ export class EntityManagerOverlay {
         })
     )
     await this.em.save(nextIds)
-    await this.afterDbUpdte(this.em)
+    await this.afterDbUpdate(this.em)
   }
 }