fix: check ipni advertisement during upload

Author: SgtPooki

src/common/upload-flow.ts

@@ -248,6 +248,7 @@ export async function performUpload(
   const uploadResult = await executeUpload(synapseService, carData, rootCid, {
     logger,
     contextId: `${contextType}-${Date.now()}`,
+    ipniValidation: { enabled: false },
     callbacks: {
       onUploadComplete: () => {
         spinner?.message('Upload complete, adding to data set...')

src/core/upload/index.ts

@@ -11,6 +11,10 @@ import {
   validatePaymentRequirements,
 } from '../payments/index.js'
 import { isSessionKeyMode, type SynapseService } from '../synapse/index.js'
+import {
+  type ValidateIPNIAdvertisementOptions,
+  validateIPNIAdvertisement,
+} from '../utils/validate-ipni-advertisement.js'
 import { type SynapseUploadResult, uploadToSynapse } from './synapse.js'
 
 export type { SynapseUploadOptions, SynapseUploadResult } from './synapse.js'
@@ -179,13 +183,30 @@ export interface UploadExecutionOptions {
   callbacks?: UploadCallbacks
   /** Optional metadata to associate with the upload. */
   metadata?: Record<string, string>
+  /**
+   * Optional IPNI validation behaviour. When enabled (default), the upload flow will wait for the IPFS Root CID to be announced to IPNI.
+   */
+  ipniValidation?: {
+    /**
+     * Enable the IPNI validation wait.
+     *
+     * @default: true
+     */
+    enabled?: boolean
+  } & ValidateIPNIAdvertisementOptions
 }
 
 export interface UploadExecutionResult extends SynapseUploadResult {
   /** Active network derived from the Synapse instance. */
   network: string
   /** Transaction hash from the piece-addition step (if available). */
   transactionHash?: string | undefined
+  /**
+   * True if the IPFS Root CID was observed on filecoinpin.contact (IPNI).
+   *
+   * You should block any displaying, or attempting to access, of IPFS download URLs unless the IPNI validation is successful.
+   */
+  ipniValidated: boolean
 }
 
 /**
@@ -200,10 +221,24 @@ export async function executeUpload(
 ): Promise<UploadExecutionResult> {
   const { logger, contextId, callbacks } = options
   let transactionHash: string | undefined
+  let ipniValidationPromise: Promise<boolean> | undefined
 
   const mergedCallbacks: UploadCallbacks = {
     onUploadComplete: (pieceCid) => {
       callbacks?.onUploadComplete?.(pieceCid)
+      // Begin IPNI validation as soon as the upload completes
+      if (options.ipniValidation?.enabled !== false && ipniValidationPromise == null) {
+        try {
+          const { enabled: _enabled, ...rest } = options.ipniValidation ?? {}
+          ipniValidationPromise = validateIPNIAdvertisement(rootCid, {
+            ...rest,
+            logger,
+          })
+        } catch (error) {
+          logger.error({ error }, 'Could not begin IPNI advertisement validation')
+          ipniValidationPromise = Promise.resolve(false)
+        }
+      }
     },
     onPieceAdded: (transaction) => {
       if (transaction?.hash) {
@@ -228,10 +263,24 @@ export async function executeUpload(
 
   const uploadResult = await uploadToSynapse(synapseService, carData, rootCid, logger, uploadOptions)
 
+  // Optionally validate IPNI advertisement of the root CID before returning
+  let ipniValidated = false
+  if (options.ipniValidation?.enabled !== false) {
+    try {
+      if (ipniValidationPromise) {
+        ipniValidated = await ipniValidationPromise
+      }
+    } catch (error) {
+      logger.error({ error }, 'Could not validate IPNI advertisement')
+      ipniValidated = false
+    }
+  }
+
   const result: UploadExecutionResult = {
     ...uploadResult,
     network: synapseService.synapse.getNetwork(),
     transactionHash,
+    ipniValidated,
   }
 
   return result

src/core/utils/index.ts

@@ -1 +1,2 @@
 export * from './format.js'
+export * from './validate-ipni-advertisement.js'

src/core/utils/validate-ipni-advertisement.ts

@@ -0,0 +1,126 @@
+import type { CID } from 'multiformats/cid'
+import type { Logger } from 'pino'
+
+export interface ValidateIPNIAdvertisementOptions {
+  /**
+   * maximum number of attempts
+   *
+   * @default: 10
+   */
+  maxAttempts?: number | undefined
+
+  /**
+   * delay between attempts in milliseconds
+   *
+   * @default: 5000
+   */
+  delayMs?: number | undefined
+
+  /**
+   * Abort signal
+   *
+   * @default: undefined
+   */
+  signal?: AbortSignal | undefined
+
+  /**
+   * Logger instance
+   *
+   * @default: undefined
+   */
+  logger?: Logger | undefined
+
+  /**
+   * Callback for progress updates
+   *
+   * @default: undefined
+   */
+  onProgress?: (
+    event:
+      | { type: 'retryUpdate'; data: { retryCount: number } }
+      | { type: 'complete'; data: { result: boolean; retryCount: number } }
+  ) => undefined | undefined
+}
+
+/**
+ * Check if the SP has announced the IPFS root CID to IPNI.
+ *
+ * This should not be called until you receive confirmation from the SP that the piece has been parked, i.e. `onPieceAdded` in the `synapse.storage.upload` callbacks.
+ *
+ * @param ipfsRootCid - The IPFS root CID to check
+ * @param options - Options for the check
+ * @returns True if the IPNI announce succeeded, false otherwise
+ */
+export async function validateIPNIAdvertisement(
+  ipfsRootCid: CID,
+  options?: ValidateIPNIAdvertisementOptions
+): Promise<boolean> {
+  const delayMs = options?.delayMs ?? 5000
+  const maxAttempts = options?.maxAttempts ?? 10
+
+  return new Promise<boolean>((resolve, reject) => {
+    let retryCount = 0
+    const check = async (): Promise<void> => {
+      try {
+        if (options?.signal?.aborted) {
+          reject(new Error('Check IPNI announce aborted'))
+          return
+        }
+        options?.logger?.info(
+          {
+            event: 'check-ipni-announce',
+            ipfsRootCid: ipfsRootCid.toString(),
+          },
+          'Checking IPNI for announcement of IPFS Root CID "%s"',
+          ipfsRootCid.toString()
+        )
+        const fetchOptions: RequestInit = {}
+        if (options?.signal) {
+          fetchOptions.signal = options?.signal
+        }
+        try {
+          options?.onProgress?.({ type: 'retryUpdate', data: { retryCount } })
+        } catch (error) {
+          options?.logger?.error({ error }, 'Error in consumer onProgress callback for retryUpdate event')
+        }
+
+        const response = await fetch(`https://filecoinpin.contact/cid/${ipfsRootCid}`, fetchOptions)
+        if (response.ok) {
+          try {
+            options?.onProgress?.({ type: 'complete', data: { result: true, retryCount } })
+          } catch (error) {
+            options?.logger?.error({ error }, 'Error in consumer onProgress callback for complete event')
+          }
+          resolve(true)
+          return
+        }
+        if (++retryCount < maxAttempts) {
+          options?.logger?.info(
+            { retryCount, maxAttempts },
+            'IPFS Root CID "%s" not announced to IPNI yet (%d/%d). Retrying in %dms...',
+            ipfsRootCid.toString(),
+            retryCount,
+            maxAttempts,
+            delayMs
+          )
+          await new Promise((resolve) => setTimeout(resolve, delayMs))
+          await check()
+        } else {
+          const msg = `IPFS root CID "${ipfsRootCid.toString()}" not announced to IPNI after ${maxAttempts} attempts`
+          const error = new Error(msg)
+          options?.logger?.error({ error }, msg)
+          try {
+            options?.onProgress?.({ type: 'complete', data: { result: false, retryCount } })
+          } catch (error) {
+            options?.logger?.error({ error }, 'Error in consumer onProgress callback for complete event')
+          }
+          reject(error)
+        }
+      } catch (error) {
+        reject(error)
+      }
+    }
+
+    check().catch(reject)
+  })
+}