feat: new core piece url retriever

.github/knip.jsonc

@@ -45,7 +45,7 @@
         "src/index.ts",
         "src/sp/index.ts",
         "src/chains.ts",
-        "src/piece.ts",
+        "src/piece/index.ts",
         "src/usdfc.ts",
         "src/abis/index.ts",
         "src/auction/index.ts",

examples/cli/src/commands/upload-dataset.ts

@@ -55,6 +55,7 @@ export const uploadDataset: Command = command(
       await SP.findPiece({
         pieceCid,
         serviceURL: provider.pdp.serviceURL,
+        retry: true,
       })
 
       const rsp = await SP.createDataSetAndAddPieces(client, {

examples/cli/src/utils.ts

@@ -1,7 +1,7 @@
 import * as p from '@clack/prompts'
 import type { Chain } from '@filoz/synapse-core/chains'
 import { getPieces } from '@filoz/synapse-core/pdp-verifier'
-import { getPDPProviders } from '@filoz/synapse-core/sp-registry'
+import { getApprovedPDPProviders } from '@filoz/synapse-core/sp-registry'
 import {
   getPdpDataSets,
   type PdpDataSet,
@@ -111,7 +111,7 @@ export async function selectProvider(
   spinner.start(`Fetching providers...`)
 
   try {
-    const { providers } = await getPDPProviders(client)
+    const providers = await getApprovedPDPProviders(client)
     spinner.stop(`Providers fetched.`)
 
     if (providers.length === 0) {

packages/synapse-core/package.json

@@ -82,8 +82,8 @@
       "default": "./dist/src/errors/index.js"
     },
     "./piece": {
-      "types": "./dist/src/piece.d.ts",
-      "default": "./dist/src/piece.js"
+      "types": "./dist/src/piece/index.d.ts",
+      "default": "./dist/src/piece/index.js"
     },
     "./utils": {
       "types": "./dist/src/utils/index.d.ts",
@@ -141,7 +141,7 @@
         "./dist/src/errors/index"
       ],
       "piece": [
-        "./dist/src/piece"
+        "./dist/src/piece/index"
       ],
       "utils": [
         "./dist/src/utils/index"
@@ -235,7 +235,9 @@
     "iso-web": "^2.1.0",
     "multiformats": "^13.4.1",
     "ox": "catalog:",
+    "p-locate": "^7.0.0",
     "p-retry": "^7.1.0",
+    "p-some": "^7.0.0",
     "zod": "catalog:"
   },
   "devDependencies": {

packages/synapse-core/src/chains.ts

@@ -332,6 +332,7 @@ export function getChain(id?: number): Chain {
 
 /**
  * Convert a viem chain to a filecoin chain.
+ *
  * @param chain - The viem chain.
  * @returns The filecoin chain.
  * @throws Errors {@link asChain.ErrorType}

packages/synapse-core/src/errors/index.ts

@@ -13,3 +13,4 @@ export * from './chains.ts'
 export * from './erc20.ts'
 export * from './pay.ts'
 export * from './pdp.ts'
+export * from './piece.ts'

packages/synapse-core/src/errors/piece.ts

@@ -0,0 +1,19 @@
+import { isSynapseError, SynapseError, type SynapseErrorOptions } from './base.ts'
+
+export class InvalidPieceCIDError extends SynapseError {
+  override name: 'InvalidPieceCIDError' = 'InvalidPieceCIDError'
+
+  constructor(input: unknown, options?: SynapseErrorOptions) {
+    let msg = 'Invalid piece CID'
+    if (typeof input === 'object' && input != null && 'toString' in input && typeof input.toString === 'function') {
+      msg = `Invalid piece CID: ${input.toString()}`
+    } else if (typeof input === 'string') {
+      msg = `Invalid piece CID: ${input}`
+    }
+    super(msg, options)
+  }
+
+  static override is(value: unknown): value is InvalidPieceCIDError {
+    return isSynapseError(value) && value.name === 'InvalidPieceCIDError'
+  }
+}

packages/synapse-core/src/index.ts

@@ -17,7 +17,7 @@ export * as erc20 from './erc20/index.ts'
 export * as errors from './errors/index.ts'
 
 export * as pay from './pay/index.ts'
-export * as piece from './piece.ts'
+export * as piece from './piece/index.ts'
 export * as sessionKey from './session-key/index.ts'
 export * as sp from './sp/index.ts'
 export * as spRegistry from './sp-registry/index.ts'

packages/synapse-core/src/mocks/jsonrpc/index.ts

@@ -16,7 +16,7 @@ import {
   stringToHex,
   toHex,
 } from 'viem'
-import * as Piece from '../../piece.ts'
+import * as Piece from '../../piece/piece.ts'
 import { TIME_CONSTANTS } from '../../utils/constants.ts'
 import { ADDRESSES } from './constants.ts'
 import { endorsementsCallHandler } from './endorsements.ts'

packages/synapse-core/src/pdp-verifier/get-active-pieces.ts

@@ -11,7 +11,7 @@ import type {
 import { readContract } from 'viem/actions'
 import type { pdpVerifierAbi } from '../abis/generated.ts'
 import { asChain } from '../chains.ts'
-import { hexToPieceCID, type PieceCID } from '../piece.ts'
+import { hexToPieceCID, type PieceCID } from '../piece/piece.ts'
 import type { ActionCallChain } from '../types.ts'
 
 export namespace getActivePieces {

packages/synapse-core/src/pdp-verifier/get-pieces.ts

@@ -2,7 +2,7 @@ import type { Simplify } from 'type-fest'
 import type { Address, Chain, Client, ReadContractErrorType, Transport } from 'viem'
 import { multicall } from 'viem/actions'
 import { asChain } from '../chains.ts'
-import { hexToPieceCID } from '../piece.ts'
+import { hexToPieceCID } from '../piece/piece.ts'
 import { metadataArrayToObject } from '../utils/metadata.ts'
 import { createPieceUrl } from '../utils/piece-url.ts'
 import { getAllPieceMetadataCall } from '../warm-storage/get-all-piece-metadata.ts'

packages/synapse-core/src/piece/download.ts

@@ -0,0 +1,138 @@
+import { type AbortError, HttpError, type NetworkError, request, type TimeoutError } from 'iso-web/http'
+import { DownloadPieceError } from '../errors/pdp.ts'
+import { InvalidPieceCIDError } from '../errors/piece.ts'
+import { asPieceCID, createPieceCIDStream, type PieceCID } from './piece.ts'
+
+export namespace download {
+  export type OptionsType = {
+    url: string
+  }
+  export type ReturnType = Uint8Array
+  export type ErrorType = DownloadPieceError | TimeoutError | NetworkError | AbortError
+}
+
+/**
+ * Download a piece from a URL.
+ *
+ * @param options - {@link download.OptionsType}
+ * @returns Data {@link download.ReturnType}
+ * @throws Errors {@link download.ErrorType}
+ */
+export async function download(options: download.OptionsType): Promise<download.ReturnType> {
+  const response = await request.get(options.url)
+  if (response.error) {
+    if (HttpError.is(response.error)) {
+      throw new DownloadPieceError(await response.error.response.text())
+    }
+    throw response.error
+  }
+  return new Uint8Array(await response.result.arrayBuffer())
+}
+
+export namespace downloadAndValidate {
+  export type OptionsType = {
+    url: string
+    expectedPieceCid: string | PieceCID
+  }
+  export type ReturnType = Uint8Array
+  export type ErrorType = DownloadPieceError | TimeoutError | NetworkError | AbortError | InvalidPieceCIDError
+}
+
+/**
+ * Download data from a URL, validate its PieceCID, and return as Uint8Array
+ *
+ * This function:
+ * 1. Downloads data from the URL
+ * 2. Calculates PieceCID during streaming
+ * 3. Collects all chunks into a Uint8Array
+ * 4. Validates the calculated PieceCID matches the expected value
+ *
+ * @param options - {@link downloadAndValidate.OptionsType}
+ * @returns Data {@link downloadAndValidate.ReturnType}
+ * @throws Errors {@link downloadAndValidate.ErrorType}
+ * @example
+ * ```ts
+ * import * as Piece from '@filoz/synapse-core/piece'
+ * const data = await Piece.downloadAndValidate({
+ *   url: 'https://example.com/piece',
+ *   expectedPieceCid: 'bafkzcib...',
+ * })
+ * console.log(data)
+ * ```
+ */
+export async function downloadAndValidate(options: downloadAndValidate.OptionsType): Promise<Uint8Array> {
+  const { url, expectedPieceCid } = options
+
+  // Parse and validate the expected PieceCID
+  const parsedPieceCid = asPieceCID(expectedPieceCid)
+  if (parsedPieceCid == null) {
+    throw new InvalidPieceCIDError(expectedPieceCid)
+  }
+
+  const rsp = await request.get(url)
+  if (rsp.error) {
+    if (HttpError.is(rsp.error)) {
+      throw new DownloadPieceError(await rsp.error.response.text())
+    }
+    throw rsp.error
+  }
+
+  if (rsp.result.body == null) {
+    throw new DownloadPieceError('Response body is null')
+  }
+
+  // Create PieceCID calculation stream
+  const { stream: pieceCidStream, getPieceCID } = createPieceCIDStream()
+
+  // Create a stream that collects all chunks into an array
+  const chunks: Uint8Array[] = []
+  const collectStream = new TransformStream<Uint8Array, Uint8Array>({
+    transform(chunk: Uint8Array, controller: TransformStreamDefaultController<Uint8Array>) {
+      chunks.push(chunk)
+      controller.enqueue(chunk)
+    },
+  })
+
+  // Pipe the response through both streams
+  const pipelineStream = rsp.result.body.pipeThrough(pieceCidStream).pipeThrough(collectStream)
+
+  // Consume the stream to completion
+  const reader = pipelineStream.getReader()
+  try {
+    while (true) {
+      const { done } = await reader.read()
+      if (done) break
+    }
+  } finally {
+    reader.releaseLock()
+  }
+
+  if (chunks.length === 0) {
+    throw new DownloadPieceError('Response body is empty')
+  }
+
+  // Get the calculated PieceCID
+  const calculatedPieceCid = getPieceCID()
+
+  if (calculatedPieceCid == null) {
+    throw new DownloadPieceError('Failed to calculate PieceCID from stream')
+  }
+
+  // Verify the PieceCID
+  if (calculatedPieceCid.toString() !== parsedPieceCid.toString()) {
+    throw new DownloadPieceError(
+      `PieceCID verification failed. Expected: ${String(parsedPieceCid)}, Got: ${String(calculatedPieceCid)}`
+    )
+  }
+
+  // Combine all chunks into a single Uint8Array
+  const totalLength = chunks.reduce((acc, chunk) => acc + chunk.length, 0)
+  const result = new Uint8Array(totalLength)
+  let offset = 0
+  for (const chunk of chunks) {
+    result.set(chunk, offset)
+    offset += chunk.length
+  }
+
+  return result
+}

packages/synapse-core/src/piece/index.ts

@@ -0,0 +1,4 @@
+export * from '../utils/piece-url.ts'
+export * from './download.ts'
+export * from './piece.ts'
+export * from './resolve-piece-url.ts'