fix: upgrade filecoin api with content store to rely on roundabout

filecoin/functions/handle-filecoin-submit-message.js

@@ -3,7 +3,7 @@ import * as Sentry from '@sentry/serverless'
 import * as storefrontEvents from '@web3-storage/filecoin-api/storefront/events'
 
 import { createPieceTable } from '../store/piece.js'
-import { createDataStore, composeDataStoresWithOrderedStream } from '../store/data.js'
+import { useContentStore } from '../store/content.js'
 import { decodeMessage } from '../queue/filecoin-submit-queue.js'
 import { mustGetEnv } from './utils.js'
 
@@ -14,7 +14,6 @@ Sentry.AWSLambda.init({
 })
 
 const AWS_REGION = process.env.AWS_REGION || 'us-west-2'
-const R2_REGION = process.env.R2_REGION || 'auto'
 
 /**
  * Get EventRecord from the SQS Event triggering the handler.
@@ -38,24 +37,12 @@ async function handleFilecoinSubmitMessage (sqsEvent) {
   // create context
   const {
     pieceTableName,
-    s3BucketName,
-    r2BucketName,
-    r2BucketEndpoint,
-    r2BucketAccessKeyId,
-    r2BucketSecretAccessKey
+    contentStoreHttpEndpoint
+
   } = getEnv()
   const context = {
     pieceStore: createPieceTable(AWS_REGION, pieceTableName),
-    dataStore: composeDataStoresWithOrderedStream(
-      createDataStore(AWS_REGION, s3BucketName),
-      createDataStore(R2_REGION, r2BucketName, {
-        endpoint: r2BucketEndpoint,
-        credentials: {
-          accessKeyId: r2BucketAccessKeyId,
-          secretAccessKey: r2BucketSecretAccessKey,
-        },
-      })
-    )
+    contentStore: useContentStore(contentStoreHttpEndpoint)
   }
 
   const { ok, error } = await storefrontEvents.handleFilecoinSubmitMessage(context, record)
@@ -78,12 +65,7 @@ async function handleFilecoinSubmitMessage (sqsEvent) {
 function getEnv () {
   return {
     pieceTableName: mustGetEnv('PIECE_TABLE_NAME'),
-    // carpark buckets - CAR file bytes may be found here with keys like {cid}/{cid}.car
-    s3BucketName: mustGetEnv('STORE_BUCKET_NAME'),
-    r2BucketName: mustGetEnv('R2_CARPARK_BUCKET_NAME'),
-    r2BucketEndpoint: mustGetEnv('R2_ENDPOINT'),
-    r2BucketAccessKeyId: mustGetEnv('R2_ACCESS_KEY_ID'),
-    r2BucketSecretAccessKey: mustGetEnv('R2_SECRET_ACCESS_KEY'),
+    contentStoreHttpEndpoint: new URL(mustGetEnv('CONTENT_STORE_HTTP_ENDPOINT'))
   }
 }
 

filecoin/package.json

@@ -17,8 +17,8 @@
     "@ucanto/principal": "^9.0.1",
     "@ucanto/transport": "^9.1.1",
     "@web3-storage/data-segment": "^5.1.0",
-    "@web3-storage/filecoin-api": "^5.0.0",
-    "@web3-storage/filecoin-client": "^3.3.1",
+    "@web3-storage/filecoin-api": "^6.0.1",
+    "@web3-storage/filecoin-client": "^3.3.3",
     "fr32-sha2-256-trunc254-padded-binary-tree-multihash": "^3.3.0",
     "multiformats": "^13.1.0",
     "p-retry": "^6.2.0",
@@ -36,6 +36,7 @@
   },
   "eslintConfig": {
     "rules": {
+      "unicorn/no-useless-undefined": "off",
       "unicorn/prefer-array-flat": "off",
       "unicorn/prefer-spread": "off"
     }

filecoin/store/content.js

@@ -0,0 +1,53 @@
+import { StoreOperationFailed, RecordNotFound } from '@web3-storage/filecoin-api/errors'
+import pRetry from 'p-retry'
+
+/**
+ * @typedef {import('multiformats').UnknownLink} UnknownLink
+ * @typedef {import('@web3-storage/filecoin-api/storefront/api').ContentStore<UnknownLink, Uint8Array>} ContentStore
+ */
+
+/**
+ * @param {URL} storeHttpEndpoint
+ * @returns {ContentStore}
+ */
+export const useContentStore = (storeHttpEndpoint) => {
+  return {
+    /**
+     * Stream Blob bytes for a given CID.
+     *
+     * @param {import('@ucanto/interface').UnknownLink} cid 
+     */
+    stream: async (cid) => {
+      // create URL for the link to be fetched
+      const getUrl = new URL(`/${cid.toString()}`, storeHttpEndpoint)
+
+      // Retry a few times as it looks like R2 sometimes takes a bit to make it available
+      let res
+      try {
+        res = await pRetry((async () => {
+          const fetchRes = await fetch(getUrl, {
+            // Follow potential redirects
+            redirect: 'follow',
+          })
+          if (fetchRes.status === 404) {
+            throw new RecordNotFound(`blob ${cid.toString()} not found in store`)
+          } else if (!fetchRes.ok || !fetchRes.body) {
+              throw new StoreOperationFailed(fetchRes.statusText)
+          }
+          return fetchRes.body
+        }), { retries: 5 })
+      } catch (err) {
+        /** @type {RecordNotFound | StoreOperationFailed} */
+        // @ts-ignore
+        const error = err
+        return {
+          error
+        }
+      }
+
+      return {
+        ok: res
+      }
+    }
+  }
+}