BLUE-280: IPFS Data publisher added

archiver-config.json

@@ -57,5 +57,12 @@
     }
   ],
   "ARCHIVER_MODE": "release",
-  "DevPublicKey": ""
-}
+  "DevPublicKey": "",
+  "txDigest": {
+    "web3Storage": {
+      "root_did": "did:key:REPLACE_ME",
+      "admin_email": ""
+    },
+    "enableSavingToWeb3Storage": false
+  }
+}

src/Config.ts

@@ -94,6 +94,11 @@ export interface Config {
     syncDelay: number
     apiServerPort: number
     txCronSchedule: string
+    web3Storage: {
+      root_did: string
+      admin_email: string
+    }
+    enableSavingToWeb3Storage: boolean
   }
   workerProcessesDebugLog: boolean // To enable debug logs for worker processes managed by the main process
   restrictFirstNodeSelectionByPublicKey: boolean // The flag to pick the first node that matches the PUBLIC_KEY specified in the firstNodeInfo
@@ -183,7 +188,12 @@ let config: Config = {
     cycleDiff: 10,
     syncDelay: 20,
     apiServerPort: 8084,
-    txCronSchedule: '*/5 * * * *',
+    txCronSchedule: '* * * * *',
+    web3Storage: {
+      root_did: 'did:key:', // Should be in the format: did:key:<DID>
+      admin_email: '', // Make sure the email here is the one to which the intended Web3.Storage Account is linked to
+    },
+    enableSavingToWeb3Storage: false, // Set root_did and admin_email when you enable this
   },
   workerProcessesDebugLog: false,
   restrictFirstNodeSelectionByPublicKey: false,

src/txDigester.ts

@@ -12,7 +12,7 @@ import { resolve } from 'path'
 import * as Logger from './Logger'
 import { startSaving } from './saveConsoleOutput'
 import axios from 'axios'
-
+import * as ipfsPublisher from './txDigester/ipfsPublisher'
 const configFile = join(process.cwd(), 'archiver-config.json')
 
 const start = async (): Promise<void> => {
@@ -38,6 +38,9 @@ const start = async (): Promise<void> => {
 
   await txDigesterDB.initializeDB(config)
 
+  // Initialises the Client that publishes Transaction Digest to Web3.Storage
+  if (config.txDigest.enableSavingToWeb3Storage) ipfsPublisher.init()
+
   const ARCHIVER_STATUS_CHECK_URL = `http://${config.ARCHIVER_IP}:${config.ARCHIVER_PORT}/status`
 
   cron.schedule(config.txDigest.txCronSchedule, async () => {

src/txDigester/ipfsPublisher.ts

@@ -0,0 +1,95 @@
+// Import necessary libraries and types
+import { join } from 'path'
+import { Blob } from 'buffer'
+import { TransactionDigest } from './txDigests'
+import * as W3SClient from '@web3-storage/w3up-client'
+import { fromEmail } from '@web3-storage/did-mailto'
+import { BlobLike } from '@web3-storage/w3up-client/dist/src/types'
+
+import { overrideDefaultConfig, config } from '../Config'
+import { mainLogger } from '../Logger'
+
+const configFile = join(process.cwd(), 'archiver-config.json')
+overrideDefaultConfig(configFile)
+
+const { admin_email, root_did } = config.txDigest.web3Storage
+
+type Web3StorageRootDID = `did:${string}:${string}`
+type Web3StorageAdminEmail = `${string}@${string}`
+
+let client: W3SClient.Client
+
+// Upload a file to the specified space DID
+export const uploadDigestToIPFS = async (data: TransactionDigest): Promise<void> => {
+  try {
+    mainLogger.info(`Uploading TX Digest for Cycle Range ${data.cycleStart} to ${data.cycleEnd}`)
+    await client.setCurrentSpace(root_did as Web3StorageRootDID)
+    console.log(`Uploading Data to Root-DID: ${root_did}`)
+
+    const { cycleStart: cs, cycleEnd: ce, txCount: tc, hash: h } = data
+    const optimisedJSON = { cs, ce, tc, h }
+
+    const cid = await client.uploadFile(
+      new Blob([JSON.stringify(optimisedJSON)], { type: 'application/json' }) as BlobLike
+    )
+    mainLogger.info(`✅ Uploaded to IPFS Successfully | CID: ${cid}`)
+    console.log(
+      `✅ Tx-Digest for Cycle Range (${data.cycleStart} to ${data.cycleEnd}) @ https://w3s.link/ipfs/${cid}`
+    )
+    const storageUsed = await client.currentSpace()?.usage.get()
+    mainLogger.info(`${Number(storageUsed!.ok)} bytes used on Web3.Storage.`)
+  } catch (error) {
+    console.error('Error while Uploading to IPFS: ', error)
+  }
+}
+
+/**
+ * Checks if the email has subscribed to a plan on the Web3.Storage Portal
+ * We'll most likely be Subscribed to the Starter/Free Plan (5GB of storage)
+ */
+const isSubscriptionActive = async (account: W3SClient.Account.Account): Promise<boolean> => {
+  const plan = await account.plan.get()
+  if (!plan.ok) {
+    console.error(`❌ ${admin_email} does not have an active plan subscribed on Web3.Storage.`)
+    return false
+  }
+  return true
+}
+
+// Checks if the client machine is authorised by the admin to interact with Web3.Storage provider
+const isAuthWithEmail = async (email: Web3StorageAdminEmail): Promise<boolean> => {
+  const accounts = client.accounts()
+  return Object.prototype.hasOwnProperty.call(accounts, fromEmail(email))
+}
+
+export const init = async (): Promise<void> => {
+  try {
+    console.log('CONFIG: ')
+    console.log({ admin_email, root_did })
+    console.log('Initialising IPFS publisher...')
+    if (!admin_email || !root_did) {
+      console.error('❌ admin_email or root_did cannot be empty. Values:')
+      console.error({ admin_email, root_did })
+      return
+    }
+    client = await W3SClient.create()
+
+    console.log(`Logging into Web3.Storage with Account: ${admin_email}...`)
+    if (!isAuthWithEmail(admin_email as Web3StorageAdminEmail))
+      console.log(`⏳ Owner of ${admin_email} needs to approve the Web3.Storage Auth request to proceed.`)
+
+    const account = await client.login(admin_email as Web3StorageAdminEmail)
+    if (!isSubscriptionActive(account)) {
+      console.error(`❌ ${admin_email} does not have a payment plan on Web3.Storage. Terminating...`)
+      return
+    }
+
+    await client.setCurrentSpace(root_did as Web3StorageRootDID)
+    console.log(`✅ Login Successful! | Current Space DID: ${root_did}`)
+    return
+  } catch (error) {
+    mainLogger.error(`Error while initializing Web3Storage client:-`)
+    mainLogger.error(error)
+    return
+  }
+}

src/txDigester/txDigestFunctions.ts

@@ -2,6 +2,7 @@ import { config } from '../Config'
 import * as processedTxs from '../dbstore/processedTxs'
 import * as txDigest from './txDigests'
 import * as Crypto from '../Crypto'
+import { uploadDigestToIPFS } from './ipfsPublisher'
 
 let lastProcessedTxDigest: txDigest.TransactionDigest = null
 
@@ -80,6 +81,9 @@ export const processAndInsertTxDigests = async (
 
     try {
       txDigest.insertTransactionDigest(txDigestObj)
+      if (config.txDigest.enableSavingToWeb3Storage) {
+        await uploadDigestToIPFS(txDigestObj)
+      }
     } catch (e) {
       console.error('Failed to insert txDigestObj: ', txDigestObj)
       console.error(e)