feat: add the EIPFS infra for the blob protocol (#361)

- adds the EIPFS multihashes queue;
- adds the EIPFS blocks CAR positions table.

.env.tpl

@@ -13,6 +13,8 @@
 
 EIPFS_INDEXER_SQS_ARN = 'arn:aws:sqs:us-west-2:505595374361:staging-ep-indexer-topic'
 EIPFS_INDEXER_SQS_URL = 'https://sqs.us-west-2.amazonaws.com/505595374361/staging-ep-indexer-topic'
+EIPFS_MULTIHASHES_SQS_ARN = 'arn:aws:sqs:us-west-2:505595374361:staging-ep-multihashes-topic'
+EIPFS_BLOCKS_CAR_POSITION_TABLE_ARN = 'arn:aws:dynamodb:us-west-2:505595374361:table/staging-ep-v1-blocks-cars-position'
 
 PROVIDERS = ''
 UPLOAD_API_DID = ''

README.md

@@ -199,6 +199,14 @@ AWS ARN for Elastic IPFS SQS indexer used to request Elastic IPFS to index given
 
 AWS URL for Elastic IPFS SQS indexer used to request Elastic IPFS to index given CAR files.
 
+#### `EIPFS_MULTIHASHES_SQS_ARN`
+
+AWS ARN for Elastic IPFS SQS multihashes used to enqueue multihashes for indexed CAR files.
+
+#### `EIPFS_BLOCKS_CAR_POSITION_TABLE_ARN`
+
+AWS ARN for Elastic IPFS DynamoDB table used to store blocks and positions for indexed CAR files.
+
 #### `POSTMARK_TOKEN`
 
 Postmark API token, which is used by the email verification system to send emails.

stacks/config.js

@@ -176,6 +176,8 @@ export function getEnv() {
     AGGREGATOR_URL: mustGetEnv('AGGREGATOR_URL'),
     CONTENT_CLAIMS_DID: mustGetEnv('CONTENT_CLAIMS_DID'),
     CONTENT_CLAIMS_URL: mustGetEnv('CONTENT_CLAIMS_URL'),
+    EIPFS_MULTIHASHES_SQS_ARN: mustGetEnv('EIPFS_MULTIHASHES_SQS_ARN'),
+    EIPFS_BLOCKS_CAR_POSITION_TABLE_ARN: mustGetEnv('EIPFS_BLOCKS_CAR_POSITION_TABLE_ARN'),
     // Not required
     STOREFRONT_PROOF: process.env.STOREFRONT_PROOF ?? '',
     CONTENT_CLAIMS_PROOF: process.env.CONTENT_CLAIMS_PROOF ?? '',

stacks/upload-api-stack.js

@@ -3,8 +3,12 @@ import {
   Config,
   Function,
   Queue,
+  Table,
   use
 } from 'sst/constructs'
+import * as sqs from 'aws-cdk-lib/aws-sqs'
+import * as dynamodb from 'aws-cdk-lib/aws-dynamodb'
+
 import { StartingPosition } from 'aws-cdk-lib/aws-lambda'
 import { UploadDbStack } from './upload-db-stack.js'
 import { BillingDbStack } from './billing-db-stack.js'
@@ -18,7 +22,7 @@ import { getCustomDomain, getApiPackageJson, getGitInfo, setupSentry, getEnv, ge
  * @param {import('sst/constructs').StackContext} properties
  */
 export function UploadApiStack({ stack, app }) {
-  const { AGGREGATOR_DID } = getEnv()
+  const { AGGREGATOR_DID, EIPFS_MULTIHASHES_SQS_ARN, EIPFS_BLOCKS_CAR_POSITION_TABLE_ARN } = getEnv()
 
   // Setup app monitoring with Sentry
   setupSentry(app, stack)
@@ -30,6 +34,28 @@ export function UploadApiStack({ stack, app }) {
   const { customerTable, spaceDiffTable, spaceSnapshotTable, stripeSecretKey } = use(BillingDbStack)
   const { pieceOfferQueue, filecoinSubmitQueue } = use(FilecoinStack)
 
+  // Blob protocol
+  // Elastic IPFS event for multihashes
+  const multihashesQueue = new Queue(stack, 'multihashes-topic-queue', {
+    cdk: {
+      queue: sqs.Queue.fromQueueArn(
+        stack,
+        'multihashes-topic',
+        EIPFS_MULTIHASHES_SQS_ARN
+      ),
+    },
+  })
+
+  const blocksCarPositionTable = new Table(stack, 'blocks-car-position-table', {
+    cdk: {
+      table: dynamodb.Table.fromTableArn(
+        stack,
+        'blocks-car-position',
+        EIPFS_BLOCKS_CAR_POSITION_TABLE_ARN
+      ),
+    },
+  })
+
   // Setup API
   const customDomain = getCustomDomain(stack.stage, process.env.HOSTED_ZONE)
   const pkg = getApiPackageJson()
@@ -62,7 +88,8 @@ export function UploadApiStack({ stack, app }) {
           workflowBucket,
           ucanStream,
           pieceOfferQueue,
-          filecoinSubmitQueue
+          filecoinSubmitQueue,
+          multihashesQueue,
         ],
         environment: {
           DID: process.env.UPLOAD_API_DID ?? '',
@@ -89,6 +116,8 @@ export function UploadApiStack({ stack, app }) {
           PIECE_TABLE_NAME: pieceTable.tableName,
           PIECE_OFFER_QUEUE_URL: pieceOfferQueue.queueUrl,
           FILECOIN_SUBMIT_QUEUE_URL: filecoinSubmitQueue.queueUrl,
+          MULTIHASHES_QUEUE_URL: multihashesQueue.queueUrl,
+          BLOCKS_CAR_POSITION_TABLE_NAME: blocksCarPositionTable.tableName,
           NAME: pkg.name,
           VERSION: pkg.version,
           COMMIT: git.commmit,