Merge pull request #4214 from thematters/feat/ipns-trigger

IPNS trigger

Author: gitwoz

.env.example

@@ -23,6 +23,7 @@ MATTERS_AWS_ARCHIVE_USER_QUEUE_URL="https://sqs.ap-southeast-1.amazonaws.com/903
 MATTERS_AWS_LIKECOIN_LIKE_QUEUE_URL="https://sqs.ap-southeast-1.amazonaws.com/903380195283/likecoin-like-dev.fifo"
 MATTERS_AWS_LIKECOIN_SEND_PV_QUEUE_URL=""
 MATTERS_AWS_LIKECOIN_UPDATE_CIVIC_LIKER_CACHE_QUEUE_URL="https://sqs.ap-southeast-1.amazonaws.com/903380195283/likecoin-update-civic-liker-cache-dev"
+MATTERS_AWS_IPNS_USER_PUBLICATION_QUEUE_URL="https://sqs.ap-southeast-1.amazonaws.com/903380195283/ipns-user-publication-dev"
 
 MATTERS_PG_HOST=db
 MATTERS_PG_USER=postgres

db/migrations/20241105155724_alter_user_ens.js

@@ -0,0 +1,17 @@
+const table = 'user'
+const ensNameColumn = 'ens_name'
+const ensNameUpdatedAtColumn = 'ens_name_updated_at'
+
+exports.up = async (knex) => {
+  await knex.schema.table(table, (t) => {
+    t.string(ensNameColumn)
+    t.timestamp(ensNameUpdatedAtColumn)
+  })
+}
+
+exports.down = async (knex) => {
+  await knex.schema.table(table, (t) => {
+    t.dropColumn(ensNameColumn)
+    t.dropColumn(ensNameUpdatedAtColumn)
+  })
+}

src/common/enums/payment.ts

@@ -1,4 +1,11 @@
-import { optimism, optimismSepolia, polygon, polygonMumbai } from 'viem/chains'
+import {
+  mainnet,
+  optimism,
+  optimismSepolia,
+  polygon,
+  polygonMumbai,
+  sepolia,
+} from 'viem/chains'
 
 import { environment, isProd } from 'common/environment'
 import { GQLChain } from 'definitions'
@@ -66,6 +73,8 @@ export const BLOCKCHAIN_CHAINID = {
 } as const
 
 export const BLOCKCHAIN_RPC: { [chainId: string]: string } = {
+  [mainnet.id]: `https://eth-mainnet.g.alchemy.com/v2/${environment.alchemyApiKey}`,
+  [sepolia.id]: `https://eth-sepolia.g.alchemy.com/v2/${environment.alchemyApiKey}`,
   [polygon.id]: `https://polygon-mainnet.g.alchemy.com/v2/${environment.alchemyApiKey}`,
   [optimism.id]: `https://opt-mainnet.g.alchemy.com/v2/${environment.alchemyApiKey}`,
   [optimismSepolia.id]: `https://opt-sepolia.g.alchemy.com/v2/${environment.alchemyApiKey}`,

src/common/enums/sqs.ts

@@ -14,4 +14,7 @@ export const QUEUE_URL = {
   // sendmail
   mail: environment?.awsMailQueueUrl,
   expressMail: environment?.awsExpressMailQueueUrl,
+
+  // IPNS
+  ipnsUserPublication: environment?.awsIpnsUserPublicationQueueUrl,
 } as const

src/common/environment.ts

@@ -50,6 +50,8 @@ export const environment = {
     process.env.MATTERS_AWS_LIKECOIN_SEND_PV_QUEUE_URL || '',
   awsLikecoinUpdateCivicLikerCache:
     process.env.MATTERS_AWS_LIKECOIN_UPDATE_CIVIC_LIKER_CACHE_QUEUE_URL || '',
+  awsIpnsUserPublicationQueueUrl:
+    process.env.MATTERS_AWS_IPNS_USER_PUBLICATION_QUEUE_URL || '',
   tsQiServerUrl: process.env.MATTERS_TSQI_SERVER_URL || '',
   awsCloudFrontEndpoint: process.env.MATTERS_AWS_CLOUD_FRONT_ENDPOINT,
   cloudflareAccountId: process.env.MATTERS_CLOUDFLARE_ACCOUNT_ID,

src/connectors/articleService.ts

@@ -40,6 +40,7 @@ import {
   USER_ACTION,
   USER_STATE,
   NODE_TYPES,
+  QUEUE_URL,
 } from 'common/enums'
 import { environment } from 'common/environment'
 import {
@@ -922,20 +923,29 @@ export class ArticleService extends BaseService<Article> {
     throw new NetworkError('failed publishToIPFS')
   }
 
-  // DEPRECATED, To Be Deleted
-  //  moved to IPNS-Listener
   public publishFeedToIPNS = async ({ userName }: { userName: string }) => {
     const userService = new UserService(this.connections)
 
     try {
+      // skip if no ENS name
+      const ensName = await userService.findEnsName(userName)
+      if (!ensName) {
+        return
+      }
+
       const ipnsKeyRec = await userService.findOrCreateIPNSKey(userName)
       if (!ipnsKeyRec) {
-        // cannot do anything if no ipns key
+        // cannot do anything if no IPNS key
         logger.error('create IPNS key ERROR: %o', ipnsKeyRec)
         return
       }
+
+      this.aws.sqsSendMessage({
+        messageBody: { userName, useMattersIPNS: true },
+        queueUrl: QUEUE_URL.ipnsUserPublication,
+      })
     } catch (error) {
-      logger.error('create IPNS key ERROR: %o', error)
+      logger.error('publishFeedToIPNS ERROR: %o', error)
       return
     }
   }

src/connectors/queue/publication.ts

@@ -74,21 +74,6 @@ export class PublicationQueue {
       }
     )
 
-  public refreshIPNSFeed = ({
-    userName,
-    numArticles = 50,
-    forceReplace,
-  }: {
-    userName: string
-    numArticles?: number
-    forceReplace?: boolean
-  }) =>
-    this.q.add(QUEUE_JOB.refreshIPNSFeed, {
-      userName,
-      numArticles,
-      forceReplace,
-    })
-
   /**
    * Consumers
    */
@@ -99,12 +84,6 @@ export class PublicationQueue {
       QUEUE_CONCURRENCY.publishArticle,
       this.handlePublishArticle
     )
-
-    this.q.process(
-      QUEUE_JOB.refreshIPNSFeed,
-      QUEUE_CONCURRENCY.refreshIPNSFeed,
-      this.handleRefreshIPNSFeed
-    )
   }
 
   /**
@@ -587,17 +566,4 @@ export class PublicationQueue {
   //     logger.error(e)
   //   }
   // }
-
-  private handleRefreshIPNSFeed: ProcessCallbackFunction<unknown> = async (
-    job // use Promise based job processing instead of `done`
-  ) => {
-    const articleService = new ArticleService(this.connections)
-    return articleService.publishFeedToIPNS(
-      job.data as {
-        userName: string
-        numArticles: number
-        forceReplace?: boolean
-      }
-    )
-  }
 }

src/connectors/userService.ts

@@ -35,7 +35,7 @@ import {
   recoverMessageAddress,
   trim,
 } from 'viem'
-import { polygon } from 'viem/chains'
+import { mainnet, polygon, sepolia } from 'viem/chains'
 
 import {
   OFFICIAL_NOTICE_EXTEND_TYPE,
@@ -75,6 +75,7 @@ import {
   AUDIT_LOG_STATUS,
   METRICS_NAMES,
   BLOCKCHAIN_RPC,
+  DAY,
 } from 'common/enums'
 import { environment, isProd } from 'common/environment'
 import {
@@ -2074,6 +2075,39 @@ export class UserService extends BaseService<User> {
     })
   }
 
+  public findEnsName = async (userName: string) => {
+    const user = await this.findByUserName(userName)
+    if (!user || !user.ethAddress) {
+      return
+    }
+
+    const now = new Date()
+    const oneDayAgo = new Date(now.getTime() - 1 * DAY)
+
+    if (
+      user.userName &&
+      user.ensNameUpdatedAt &&
+      user.ensNameUpdatedAt > oneDayAgo
+    ) {
+      return user.ensName
+    }
+
+    // Query ENS if data is stale or missing
+    const client = createPublicClient({
+      chain: isProd ? mainnet : sepolia,
+      transport: http(BLOCKCHAIN_RPC[isProd ? mainnet.id : sepolia.id]),
+    })
+    const ensName = await client.getEnsName({
+      address: user.ethAddress as `0x${string}`,
+    })
+    await this.models.update({
+      table: 'user',
+      where: { id: user.id },
+      data: { ensName, ensNameUpdatedAt: now },
+    })
+    return ensName
+  }
+
   /*********************************
    *                               *
    *        Restrictions           *

src/definitions/user.d.ts

@@ -27,6 +27,8 @@ interface UserBase {
   state: UserState
   agreeOn: string
   ethAddress: string | null
+  ensName: string | null
+  ensNameUpdatedAt: Date | null
   currency: 'HKD' | 'TWD' | 'USD' | null
   profileCover?: string
   // eslint-disable-next-line @typescript-eslint/no-explicit-any

src/mutations/user/refreshIPNSFeed.ts

@@ -1,22 +1,19 @@
 import type { GQLMutationResolvers } from 'definitions'
 
+import { QUEUE_URL } from 'common/enums'
+import { aws } from 'connectors'
+
 const resolver: GQLMutationResolvers['refreshIPNSFeed'] = async (
   _,
   { input: { userName, numArticles = 50 } },
-  {
-    dataSources: {
-      userService,
-      queues: { publicationQueue },
-    },
-  }
+  { dataSources: { userService } }
 ) => {
   const ipnsKeyRec = await userService.findOrCreateIPNSKey(userName)
 
   if (ipnsKeyRec) {
-    publicationQueue.refreshIPNSFeed({
-      userName,
-      numArticles,
-      forceReplace: true,
+    aws.sqsSendMessage({
+      messageBody: { userName, useMattersIPNS: true },
+      queueUrl: QUEUE_URL.ipnsUserPublication,
     })
   }