feat: add index provider library (#143)

* add count of payload cid state

* Update backend/bin/deal-observer-backend.js

Co-authored-by: Miroslav Bajtoš <[email protected]>

* Update backend/lib/look-up-payload-cids.js

Co-authored-by: Miroslav Bajtoš <[email protected]>

* changed testing

* moved default active deal

* changed storage function

* fmt

* add index lookup library

* fix tests

* cleanup

* Update backend/bin/deal-observer-backend.js

Co-authored-by: Julian Gruber <[email protected]>

* resolve PR threads

* fmt

* removed default value

* add secret to workflow

* add secret to workflow

* Update backend/lib/resolve-payload-cids.js

Co-authored-by: Julian Gruber <[email protected]>

* fmt

* remove glif token during deployment

* fix async test

* Update backend/lib/resolve-payload-cids.js

Co-authored-by: Julian Gruber <[email protected]>

* Update backend/lib/resolve-payload-cids.js

Co-authored-by: Julian Gruber <[email protected]>

* removed redundant async

---------

Co-authored-by: Nikolas Haimerl <[email protected]>
Co-authored-by: Miroslav Bajtoš <[email protected]>
Co-authored-by: Julian Gruber <[email protected]>

Author: 4 people

.github/workflows/ci.yml

@@ -52,6 +52,7 @@ jobs:
     env:
       DATABASE_URL: postgres://postgres:postgres@localhost:5432/spark_deal_observer
       NPM_CONFIG_WORKSPACE: backend
+      GLIF_TOKEN: ${{ secrets.GLIF_TOKEN }}
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-node@v4

backend/bin/deal-observer-backend.js

@@ -7,7 +7,7 @@ import '../lib/instrument.js'
 import { createInflux } from '../lib/telemetry.js'
 import { rpcRequest } from '../lib/rpc-service/service.js'
 import { fetchDealWithHighestActivatedEpoch, countStoredActiveDeals, observeBuiltinActorEvents } from '../lib/deal-observer.js'
-import { countStoredActiveDealsWithUnresolvedPayloadCid, resolvePayloadCids, countRevertedActiveDeals, countStoredActiveDealsWithPayloadState } from '../lib/resolve-payload-cids.js'
+import { countStoredActiveDealsWithUnresolvedPayloadCid, resolvePayloadCids, countRevertedActiveDeals, countStoredActiveDealsWithPayloadState, getPeerId } from '../lib/resolve-payload-cids.js'
 import { findAndSubmitUnsubmittedDeals, submitDealsToSparkApi } from '../lib/spark-api-submit-deals.js'
 import { payloadCidRequest } from '../lib/piece-indexer-service.js'
 /** @import {Queryable} from '@filecoin-station/deal-observer-db' */
@@ -122,17 +122,16 @@ const sparkApiSubmitDealsLoop = async (pgPool, { sparkApiBaseUrl, sparkApiToken,
 }
 
 /**
- * @param {MakeRpcRequest} makeRpcRequest
  * @param {MakePayloadCidRequest} makePayloadCidRequest
  * @param {Queryable} pgPool
  */
-export const resolvePayloadCidsLoop = async (makeRpcRequest, makePayloadCidRequest, pgPool) => {
+export const resolvePayloadCidsLoop = async (makePayloadCidRequest, pgPool) => {
   while (true) {
     const start = Date.now()
     // Maximum number of deals to resolve payload CIDs for in one loop iteration
     const maxDeals = 1000
     try {
-      const numOfPayloadCidsResolved = await resolvePayloadCids(makeRpcRequest, makePayloadCidRequest, pgPool, maxDeals)
+      const numOfPayloadCidsResolved = await resolvePayloadCids(getPeerId, makePayloadCidRequest, pgPool, maxDeals)
       const totalNumOfUnresolvedPayloadCids = await countStoredActiveDealsWithUnresolvedPayloadCid(pgPool)
       const totalNumOfDealsWithPayloadCidResolved = await countStoredActiveDealsWithPayloadState(pgPool, 'PAYLOAD_CID_RESOLVED')
       const totalNumOfDealsWithPayloadCidUnresolved = await countStoredActiveDealsWithPayloadState(pgPool, 'PAYLOAD_CID_UNRESOLVED')
@@ -169,7 +168,7 @@ export const resolvePayloadCidsLoop = async (makeRpcRequest, makePayloadCidReque
 }
 
 await Promise.all([
-  resolvePayloadCidsLoop(rpcRequest, payloadCidRequest, pgPool),
+  resolvePayloadCidsLoop(payloadCidRequest, pgPool),
   observeActorEventsLoop(rpcRequest, pgPool),
   sparkApiSubmitDealsLoop(pgPool, {
     sparkApiBaseUrl: SPARK_API_BASE_URL,

backend/lib/config.js

@@ -17,5 +17,6 @@ if (RPC_URL.includes('glif') && GLIF_TOKEN) {
 export {
   RPC_URL,
   rpcHeaders,
-  PIECE_INDEXER_URL
+  PIECE_INDEXER_URL,
+  GLIF_TOKEN
 }

backend/lib/resolve-payload-cids.js

@@ -1,7 +1,16 @@
 import { loadDeals } from './deal-observer.js'
 import * as util from 'node:util'
-import { getMinerPeerId } from './rpc-service/service.js'
 import { PayloadRetrievabilityState } from '@filecoin-station/deal-observer-db/lib/types.js'
+import debug from 'debug'
+import { GLIF_TOKEN, RPC_URL } from './config.js'
+import { ethers } from 'ethers'
+import {
+  getIndexProviderPeerId,
+  MINER_TO_PEERID_CONTRACT_ADDRESS, MINER_TO_PEERID_CONTRACT_ABI
+// @ts-ignore
+} from 'index-provider-peer-id'
+import { rpcRequest } from './rpc-service/service.js'
+import assert from 'node:assert'
 
 /** @import {Queryable} from '@filecoin-station/deal-observer-db' */
 /** @import { Static } from '@sinclair/typebox' */
@@ -11,16 +20,18 @@ const THREE_DAYS_IN_MILLISECONDS = 1000 * 60 * 60 * 24 * 3
 
 /**
  *
- * @param {import('./typings.js').MakeRpcRequest} makeRpcRequest
+ * @param {import('./typings.js').GetIndexProviderPeerId} getIndexProviderPeerId
  * @param {import('./typings.js').MakePayloadCidRequest} makePayloadCidRequest
  * @param {Queryable} pgPool
  * @param {number} maxDeals
+ * @param {number} [now] The current timestamp in milliseconds
  * @returns {Promise<number>}
  */
-export const resolvePayloadCids = async (makeRpcRequest, makePayloadCidRequest, pgPool, maxDeals, now = Date.now()) => {
+export const resolvePayloadCids = async (getIndexProviderPeerId, makePayloadCidRequest, pgPool, maxDeals, now = Date.now()) => {
   let payloadCidsResolved = 0
   for (const deal of await fetchDealsWithUnresolvedPayloadCid(pgPool, maxDeals, new Date(now - THREE_DAYS_IN_MILLISECONDS))) {
-    const minerPeerId = await getMinerPeerId(deal.miner_id, makeRpcRequest)
+    const { peerId: minerPeerId, source } = await getIndexProviderPeerId(deal.miner_id)
+    debug(`Using PeerID from ${source}.`)
     const payloadCid = await makePayloadCidRequest(minerPeerId, deal.piece_cid)
     if (payloadCid) deal.payload_cid = payloadCid
     if (!deal.payload_cid) {
@@ -106,3 +117,31 @@ async function updatePayloadCidInActiveDeal (pgPool, deal, newPayloadRetrievalSt
     throw Error(util.format('Error updating payload of deal: ', deal), { cause: error })
   }
 }
+
+function getSmartContractClient () {
+  const fetchRequest = new ethers.FetchRequest(RPC_URL)
+  assert(GLIF_TOKEN, 'GLIF_TOKEN is required')
+  fetchRequest.setHeader('Authorization', `Bearer ${GLIF_TOKEN}`)
+  const provider = new ethers.JsonRpcProvider(fetchRequest)
+  return new ethers.Contract(
+    MINER_TO_PEERID_CONTRACT_ADDRESS,
+    MINER_TO_PEERID_CONTRACT_ABI,
+    provider
+  )
+}
+const defaultSmartContractClient = getSmartContractClient()
+
+/**
+ * @param {number} minerId
+ * @param {object} [options]
+ * @param {unknown} options.smartContract
+ * @param {import('./typings.js').MakeRpcRequest} options.makeRpcRequest
+ * @returns {Promise<{ peerId: string, source: string }>}
+ */
+export const getPeerId = async (minerId, { smartContract, makeRpcRequest } = { smartContract: defaultSmartContractClient, makeRpcRequest: rpcRequest }) => {
+  return await getIndexProviderPeerId(
+  `f0${minerId}`,
+  smartContract,
+  { rpcFn: makeRpcRequest }
+  )
+}

backend/lib/rpc-service/service.js

@@ -96,28 +96,6 @@ export async function getChainHead (makeRpcRequest) {
   }
 }
 
-/**
- * @param {number} minerId
- * @param {MakeRpcRequest} makeRpcRequest
- * @returns {Promise<string>}
- */
-export async function getMinerPeerId (minerId, makeRpcRequest) {
-  /** @typedef {{
-   * PeerId: string;
-   * }} MinerInfo
-   */
-  try {
-    const params = getMinerInfoCallParams(minerId)
-    const res = /** @type {MinerInfo} */(await makeRpcRequest('Filecoin.StateMinerInfo', params))
-    if (!res || !res.PeerId) {
-      throw Error(`Failed to get peer ID for miner ${minerId}, result: ${res}`)
-    }
-    return res.PeerId
-  } catch (err) {
-    throw Error(`Failed to get peer ID for miner ${minerId}.`, { cause: err })
-  }
-}
-
 /**
    * @param {number} blockHeight
    * @param {string} eventTypeString
@@ -134,13 +112,3 @@ export function getActorEventsFilter (blockHeight, eventTypeString) {
     }
   }
 }
-
-/**
- * @param {number} minerId
- */
-export function getMinerInfoCallParams (minerId) {
-  return [
-    'f0' + minerId.toString(),
-    null
-  ]
-}

backend/lib/typings.d.ts

@@ -1,2 +1,3 @@
 export type MakeRpcRequest = (method: string, params: unknown[]) => Promise<unknown>;
 export type MakePayloadCidRequest = (providerId:string,pieceCid:string) => Promise<string|null>;
+export type GetIndexProviderPeerId = ( minerId:number ) => Promise<{ peerId: string, source: string }>;

backend/package.json

@@ -22,6 +22,8 @@
     "@sentry/node": "^9.11.0",
     "@sinclair/typebox": "^0.34.33",
     "debug": "^4.4.0",
+    "ethers": "^6.13.5",
+    "index-provider-peer-id": "^1.0.0",
     "multiformats": "^13.3.2",
     "p-retry": "^6.2.1",
     "pg": "^8.14.1",