feat: Produce item published events (#29)

package.json

@@ -19,7 +19,7 @@
     "@aws-sdk/client-sns": "^3.609.0",
     "@dcl/catalyst-contracts": "^4.4.1",
     "@dcl/platform-server-commons": "^0.0.4",
-    "@dcl/schemas": "^13.11.0",
+    "@dcl/schemas": "^13.12.0",
     "@well-known-components/env-config-provider": "^1.2.0",
     "@well-known-components/http-server": "^2.1.0",
     "@well-known-components/interfaces": "^1.4.3",

src/adapters/producers/item-published.ts

@@ -0,0 +1,120 @@
+import { Events, ItemPublishedEvent } from '@dcl/schemas'
+import { AppComponents, IEventGenerator } from '../../types'
+
+export const PAGE_SIZE = 1000
+
+const PUBLISHED_ITEMS_QUERY = `
+    query Items($updatedAt: BigInt!, $paginationId: ID) {
+      items(
+        where: {updatedAt_gte: $updatedAt, id_gt: $paginationId}
+        orderBy: id
+        orderDirection: asc
+        first: ${PAGE_SIZE}
+      ) {
+        id
+        collection{
+            id
+        }
+        blockchainId
+        creator
+        itemType
+        rarity
+        urn
+        createdAt
+        updatedAt
+        metadata {
+          wearable {
+              name
+          }
+          emote {
+              name
+              
+          }
+        }
+      }
+    }
+  `
+
+type ItemsResponse = {
+  items: {
+    id: string
+    collection: {
+      id: string
+    }
+    blockchainId: string
+    creator: string
+    itemType: string
+    rarity: string
+    urn: string
+    createdAt: number
+    updatedAt: number
+    metadata: {
+      wearable: {
+        name: string
+      }
+      emote: {
+        name: string
+      }
+    }
+  }[]
+}
+
+export async function itemPublishedProducer(
+  components: Pick<AppComponents, 'l2CollectionsSubGraph'>
+): Promise<IEventGenerator> {
+  const { l2CollectionsSubGraph } = components
+
+  async function run(updatedAt: number) {
+    const now = Date.now()
+    const produced: ItemPublishedEvent[] = []
+
+    let result: ItemsResponse
+    let paginationId = ''
+    do {
+      result = await l2CollectionsSubGraph.query<ItemsResponse>(PUBLISHED_ITEMS_QUERY, {
+        updatedAt: Math.floor(updatedAt / 1000),
+        paginationId
+      })
+
+      if (result.items.length === 0) {
+        break
+      }
+
+      for (const item of result.items) {
+        const event: ItemPublishedEvent = {
+          type: Events.Type.BLOCKCHAIN,
+          subType: Events.SubType.Blockchain.ITEM_PUBLISHED,
+          key: item.urn,
+          timestamp: item.createdAt * 1000,
+          metadata: {
+            creator: item.creator,
+            category: item.metadata.wearable ? 'wearable' : 'emote',
+            rarity: item.rarity,
+            network: 'polygon',
+            tokenId: item.id
+          }
+        }
+        produced.push(event)
+
+        paginationId = item.id
+      }
+    } while (result.items.length === PAGE_SIZE)
+
+    return {
+      event: {
+        type: Events.Type.BLOCKCHAIN,
+        subType: Events.SubType.Blockchain.ITEM_PUBLISHED
+      },
+      records: produced,
+      lastRun: now
+    }
+  }
+
+  return {
+    event: {
+      type: Events.Type.BLOCKCHAIN,
+      subType: Events.SubType.Blockchain.ITEM_PUBLISHED
+    },
+    run
+  }
+}

src/components.ts

@@ -23,6 +23,7 @@ import {
   instrumentHttpServerWithPromClientRegistry
 } from '@well-known-components/http-server'
 import { collectionCreatedProducer } from './adapters/producers/collection-created'
+import { itemPublishedProducer } from './adapters/producers/item-published'
 
 // Initialize all the components of the app
 export async function initComponents(): Promise<AppComponents> {
@@ -94,6 +95,9 @@ export async function initComponents(): Promise<AppComponents> {
       await itemSoldProducer({ config, l2CollectionsSubGraph: onlySatsumaL2CollectionsSubGraph }) // Temp fix to only listen to satsuma until the Subquid has the fix too
     )
   )
+  producerRegistry.addProducer(
+    await createProducer({ logs, database, eventPublisher }, await itemPublishedProducer({ l2CollectionsSubGraph }))
+  )
   producerRegistry.addProducer(
     await createProducer(
       { logs, database, eventPublisher },

test/unit/producers/item-published.spec.ts

@@ -0,0 +1,78 @@
+import { createConfigComponent } from '@well-known-components/env-config-provider'
+import { itemPublishedProducer } from '../../../src/adapters/producers/item-published'
+
+describe('item published producer', () => {
+  test('should work when some new published item is found', async () => {
+    const l2CollectionsSubGraph = {
+      query: jest.fn()
+    }
+    l2CollectionsSubGraph.query.mockReturnValue({
+      items: [
+        {
+          id: '0x6105f0f5ef8b28cf81e64551588d13221d4151ad-3',
+          collection: {
+            id: '0x6105f0f5ef8b28cf81e64551588d13221d4151ad'
+          },
+          blockchainId: '3',
+          creator: '0x943d99cefa84b247b679d2b7cce17c7c93e6587b',
+          itemType: 'wearable_v2',
+          rarity: 'mythic',
+          urn: 'urn:decentraland:matic:collections-v2:0x6105f0f5ef8b28cf81e64551588d13221d4151ad:3',
+          createdAt: '1623259167',
+          updatedAt: '1623355278',
+          metadata: {
+            wearable: {
+              name: 'MetaZoo Intl. BlackCat Hood'
+            },
+            emote: null
+          }
+        }
+      ]
+    })
+
+    const producer = await itemPublishedProducer({ l2CollectionsSubGraph })
+    let result = await producer.run(Date.now())
+    expect(result).toMatchObject({
+      event: {
+        type: 'blockchain',
+        subType: 'item-published'
+      },
+      records: [
+        {
+          type: 'blockchain',
+          subType: 'item-published',
+          key: 'urn:decentraland:matic:collections-v2:0x6105f0f5ef8b28cf81e64551588d13221d4151ad:3',
+          timestamp: 1623259167000,
+          metadata: {
+            creator: '0x943d99cefa84b247b679d2b7cce17c7c93e6587b',
+            category: 'wearable',
+            rarity: 'mythic',
+            network: 'polygon',
+            tokenId: '0x6105f0f5ef8b28cf81e64551588d13221d4151ad-3'
+          }
+        }
+      ],
+      lastRun: expect.anything()
+    })
+  })
+
+  test('should work when no new published items', async () => {
+    const l2CollectionsSubGraph = {
+      query: jest.fn()
+    }
+    l2CollectionsSubGraph.query.mockReturnValue({
+      items: []
+    })
+
+    const producer = await itemPublishedProducer({ l2CollectionsSubGraph })
+    let result = await producer.run(Date.now())
+    expect(result).toMatchObject({
+      event: {
+        type: 'blockchain',
+        subType: 'item-published'
+      },
+      records: [],
+      lastRun: expect.anything()
+    })
+  })
+})

yarn.lock

@@ -774,10 +774,10 @@
     "@well-known-components/interfaces" "^1.4.2"
     node-fetch "^2.7.0"
 
-"@dcl/schemas@^13.11.0":
-  version "13.11.0"
-  resolved "https://registry.yarnpkg.com/@dcl/schemas/-/schemas-13.11.0.tgz#05ba668f7a3f0d91a6d115d4afca80bd99251f6c"
-  integrity sha512-ux89szB5EZ7NcXVcOcJdzNv9/+p/7RaiukSBiy04Y7aMIrppYL5IdUUpsYiNkQ43Gig9N7XTobAInj6ZgeA6fQ==
+"@dcl/schemas@^13.12.0":
+  version "13.12.0"
+  resolved "https://registry.yarnpkg.com/@dcl/schemas/-/schemas-13.12.0.tgz#190799a538f2f009f19448fb4662ff387e3ec026"
+  integrity sha512-YHfiDSYqSRGXwiBaj5qZPzhhW7agT2bvLmrLvNZe3sFmkJTkOoFjWx5ZsG5V7JmRW6sLpJXxfKMFjcryyGU2IQ==
   dependencies:
     ajv "^8.11.0"
     ajv-errors "^3.0.0"