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

)

Today Storefront relies directly on a bucket (in S3) to read bytes to
derive PieceCID for validation. With blob protocol, we are moving to
write to R2 only 🙌🏼 therefore, we aim to make filecoin Storefront able
to read from anywhere, but also not increase the infra costs with this
change.

This PR changes Storefront to rely on Roundabout to read Blob bytes to
derive Piece CID from anywhere instead of `contentStore` being directly
hooked to a bucket. See
[usage](https://github.com/w3s-project/w3up/blob/main/packages/filecoin-api/src/storefront/events.js#L43)

Relying on Roundabout introduces a layer of indirection, but in current
implementation where this validation runs in AWS and content will likely
be read from CF, makes it cost optimal to read from Roundabout.

Closes storacha/w3up#1349

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
+      }
+    }
+  }
+}