feat(Health): delete anchored transactions from ipfsDirectoryHash col…

…lection (#891)

remove a directoryHash from Health tracking once it has been anchored to the bitcoin blockchain

References: #832, #891

src/Health/HealthController.ts

@@ -4,6 +4,7 @@ import { pick, pipeP } from 'ramda'
 
 import { childWithFileName } from 'Helpers/Logging'
 import { IPFSHashFailure, ClaimIPFSHashPair } from 'Interfaces'
+import { IPFSHashTxId } from 'Messaging/Messages'
 
 import { BlockchainInfo, WalletInfo, NetworkInfo, IPFSInfo, EstimatedSmartFeeInfo, HealthDAO } from './HealthDAO'
 import { TransactionAnchorRetryInfo, IPFSDirectoryHashDAO } from './IPFSDirectoryHashDAO'
@@ -77,6 +78,7 @@ export class HealthController {
     this.logger = childWithFileName(logger, __filename)
     this.configuration = configuration
     this.healthDAO = healthDAO
+    this.ipfsDirectoryHashDAO = ipfsDirectoryHashDAO
     this.bitcoinCore = bitcoinCore
     this.ipfs = ipfs
   }
@@ -157,6 +159,10 @@ export class HealthController {
     return transactionAnchorRetryInfo
   }
 
+  private async removeIPFSDirectoryHashByTransactionId(transactionIds: ReadonlyArray<string>): Promise<void> {
+    await this.ipfsDirectoryHashDAO.deleteByTransactionIds(transactionIds)
+  }
+
   public async updateIPFSFailures(ipfsHashFailures: ReadonlyArray<IPFSHashFailure>) {
     this.logger.debug({ ipfsHashFailures }, 'Updating IPFS Failure Count by failureType')
     await ipfsHashFailures.map(
@@ -167,6 +173,12 @@ export class HealthController {
     )
   }
 
+  public purgeIpfsDirectoryHashByTransactionIds = pipeP(
+    this.log(LogTypes.trace)('purging IPFSDirectoryHash for transactionIds'),
+    this.removeIPFSDirectoryHashByTransactionId,
+    this.log(LogTypes.trace)('purged IPFSDirectoryHash for transactionIds'),
+  )
+
   public refreshBlockchainInfo = pipeP(
     this.log(LogTypes.trace)('refreshing blockchain info'),
     this.getBlockchainInfo,

src/Health/IPFSDirectoryHashDAO.ts

@@ -17,6 +17,7 @@ export interface TransactionAnchorRetryEntry {
 
 export type TransactionAnchorRetryInfo = ReadonlyArray<TransactionAnchorRetryEntry>
 export type getTransactionAnchorRetryInfo = () => Promise<TransactionAnchorRetryInfo>
+type deleteByTransactionIds = (transactionIds: ReadonlyArray<string>) => Promise<void>
 
 export interface Dependencies {
   readonly ipfsDirectoryHashInfoCollection: Collection
@@ -56,6 +57,14 @@ export class IPFSDirectoryHashDAO {
     )
   }
 
+  readonly deleteByTransactionIds: deleteByTransactionIds = async transactionIds => {
+    await this.ipfsDirectoryHashInfoCollection.deleteMany(
+      {
+        txId: { $in: transactionIds },
+      },
+    )
+  }
+
   readonly getTransactionAnchorRetryInfo: getTransactionAnchorRetryInfo = async () => {
     const cursorArray = await this.ipfsDirectoryHashInfoCollection.aggregate([
       { $match: { attempts: { $ne: 1 } } },

src/Health/Router.ts

@@ -2,6 +2,7 @@ import * as Pino from 'pino'
 
 import { childWithFileName } from 'Helpers/Logging'
 import { IPFSHashFailure } from 'Interfaces'
+import { BlockDownloaded, IPFSHashTxId } from 'Messaging/Messages'
 import { Messaging } from 'Messaging/Messaging'
 
 import { ExchangeConfiguration } from './ExchangeConfiguration'
@@ -41,6 +42,7 @@ export class Router {
   async start() {
     await this.messaging.consume(this.exchange.getHealth, this.onGetHealth)
     await this.messaging.consumeClaimsNotDownloaded(this.onClaimsNotDownloaded)
+    await this.messaging.consumeBlockDownloaded(this.blockDownloadedConsumer)
   }
 
   onGetHealth = async () => {
@@ -67,4 +69,16 @@ export class Router {
       logger.error({ error }, 'Failed to update ipfsHashFailures on health')
     }
   }
+
+  blockDownloadedConsumer = async (blockDownloaded: BlockDownloaded): Promise<void> => {
+    const logger = this.logger.child({ method: 'blockDownloadedConsumer' })
+
+    logger.debug({ blockDownloaded }, 'Block downloaded, removing associated transactions')
+    try {
+      const transactionIds = blockDownloaded.poetBlockAnchors.map(_ => _.transactionId)
+      await this.controller.purgeIpfsDirectoryHashByTransactionIds(transactionIds)
+    } catch (error) {
+      logger.error({ error }, 'Failed to remove transactions')
+    }
+  }
 }