feat(Health): inject ipfsHashTxId into Health module (#838)

References #838, #832, #707

Author: krobi64

src/Health/ExchangeConfiguration.ts

@@ -1,4 +1,5 @@
 export interface ExchangeConfiguration {
   readonly getHealth?: string
   readonly claimsNotDownloaded?: string
+  readonly ipfsHashTxId?: string
 }

src/Health/Health.ts

@@ -51,7 +51,7 @@ export class Health {
     this.ipfsDirectoryHashInfoCollection = this.dbConnection.collection('ipfsDirectoryHashInfo')
 
     const exchangesMessaging = pick(
-      ['getHealth', 'claimsNotDownloaded'],
+      ['getHealth', 'claimsNotDownloaded', 'ipfsHashTxId'],
       this.configuration.exchanges,
     )
     this.messaging = new Messaging(this.configuration.rabbitmqUrl, exchangesMessaging)

src/Messaging/ExchangeConfiguration.ts

@@ -2,4 +2,5 @@ export interface ExchangeConfiguration {
   readonly poetAnchorDownloaded?: string
   readonly claimsDownloaded?: string
   readonly claimsNotDownloaded?: string
+  readonly ipfsHashTxId?: string
 }

src/Messaging/Messages.ts

@@ -12,6 +12,11 @@ export interface LightBlock {
   readonly height: number
 }
 
+export interface IPFSHashTxId {
+  readonly ipfsDirectoryHash: string
+  readonly txId: string
+}
+
 const PoetBlockAnchorJoiSchema = Joi.object({
   prefix: Joi.string().required(),
   version: Joi.number().required(),
@@ -29,5 +34,13 @@ const BlockDownloadedJoiSchema = Joi.object({
     .optional(),
 })
 
+const IPFSHashTxIdJoiSchema = Joi.object({
+  ipfsDirectoryHash: Joi.string().required(),
+  txId: Joi.string().required(),
+})
+
 export const isBlockDownloaded = (messageContent: any): messageContent is BlockDownloaded =>
   Joi.validate(messageContent, BlockDownloadedJoiSchema).error === null
+
+export const isIPFSHashTxId = (messageContent: any): messageContent is IPFSHashTxId =>
+  Joi.validate(messageContent, IPFSHashTxIdJoiSchema).error === null

src/Messaging/Messaging.ts

@@ -4,7 +4,7 @@ import { Connection, connect, Channel } from 'amqplib'
 import { ClaimIPFSHashPair, isClaimIPFSHashPair, IPFSHashFailure, isIPFSHashFailure } from 'Interfaces'
 
 import { ExchangeConfiguration } from './ExchangeConfiguration'
-import { BlockDownloaded, isBlockDownloaded } from './Messages'
+import { BlockDownloaded, IPFSHashTxId, isBlockDownloaded, isIPFSHashTxId } from './Messages'
 
 export class Messaging {
   private readonly connectionUrl: string
@@ -71,6 +71,21 @@ export class Messaging {
     })
   }
 
+  publishIPFSHashTxId = async (hashTxId: IPFSHashTxId) => {
+    return this.publish(this.exchanges.ipfsHashTxId, hashTxId)
+  }
+
+  consumeIPFSHashTxId = async (consume: (hashTxId: IPFSHashTxId) => void) => {
+    await this.consume(this.exchanges.ipfsHashTxId, (message: any) => {
+      const messageContent = message.content.toString()
+      const hashTxId = JSON.parse(messageContent) as unknown
+
+      if (!isIPFSHashTxId(hashTxId)) return
+
+      consume(hashTxId)
+    })
+  }
+
   publishClaimsDownloaded = async (claimIPFSHashPairs: ReadonlyArray<ClaimIPFSHashPair>) => {
     return this.publish(this.exchanges.claimsDownloaded, claimIPFSHashPairs)
   }

src/app.ts

@@ -244,6 +244,7 @@ export async function app(localVars: any = {}) {
     exchanges: {
       getHealth: configuration.exchangeGetHealth,
       claimsNotDownloaded: configuration.exchangeClaimsNotDownloaded,
+      ipfsHashTxId: configuration.exchangeIpfsHashTxId,
     },
   })