feat: add blob protocol to service

stacks/upload-api-stack.js

@@ -25,7 +25,7 @@ export function UploadApiStack({ stack, app }) {
 
   // Get references to constructs created in other stacks
   const { carparkBucket } = use(CarparkStack)
-  const { storeTable, uploadTable, delegationBucket, delegationTable, revocationTable, adminMetricsTable, spaceMetricsTable, consumerTable, subscriptionTable, rateLimitTable, pieceTable, privateKey } = use(UploadDbStack)
+  const { allocationTable, storeTable, uploadTable, delegationBucket, delegationTable, revocationTable, adminMetricsTable, spaceMetricsTable, consumerTable, subscriptionTable, rateLimitTable, pieceTable, privateKey } = use(UploadDbStack)
   const { invocationBucket, taskBucket, workflowBucket, ucanStream } = use(UcanInvocationStack)
   const { customerTable, spaceDiffTable, spaceSnapshotTable, stripeSecretKey } = use(BillingDbStack)
   const { pieceOfferQueue, filecoinSubmitQueue } = use(FilecoinStack)
@@ -41,6 +41,7 @@ export function UploadApiStack({ stack, app }) {
     defaults: {
       function: {
         permissions: [
+          allocationTable,
           storeTable,
           uploadTable,
           customerTable,
@@ -66,6 +67,7 @@ export function UploadApiStack({ stack, app }) {
         environment: {
           DID: process.env.UPLOAD_API_DID ?? '',
           AGGREGATOR_DID,
+          ALLOCATION_TABLE_NAME: allocationTable.tableName,
           STORE_TABLE_NAME: storeTable.tableName,
           STORE_BUCKET_NAME: carparkBucket.bucketName,
           UPLOAD_TABLE_NAME: uploadTable.tableName,
@@ -92,6 +94,7 @@ export function UploadApiStack({ stack, app }) {
           COMMIT: git.commmit,
           STAGE: stack.stage,
           ACCESS_SERVICE_URL: getServiceURL(stack) ?? '',
+          UPLOAD_SERVICE_URL: customDomain?.domainName ? `https://${customDomain?.domainName}` : '',
           POSTMARK_TOKEN: process.env.POSTMARK_TOKEN ?? '',
           PROVIDERS: process.env.PROVIDERS ?? '',
           R2_ACCESS_KEY_ID: process.env.R2_ACCESS_KEY_ID ?? '',

stacks/upload-db-stack.js

@@ -1,6 +1,7 @@
 import { Table, Bucket, Config } from 'sst/constructs'
 
 import {
+  allocationTableProps,
   storeTableProps,
   uploadTableProps,
   consumerTableProps,
@@ -31,6 +32,12 @@ export function UploadDbStack({ stack, app }) {
   // TODO: we should look into creating a trust layer for content claims
   const contentClaimsPrivateKey = new Config.Secret(stack, 'CONTENT_CLAIMS_PRIVATE_KEY')
 
+  /**
+   * The allocation table tracks allocated multihashes per space.
+   * Used by the blob/* service capabilities.
+   */
+  const allocationTable = new Table(stack, 'allocation', allocationTableProps)
+
   /**
    * This table takes a stored CAR and makes an entry in the store table
    * Used by the store/* service capabilities.
@@ -99,6 +106,7 @@ export function UploadDbStack({ stack, app }) {
   const spaceMetricsTable = new Table(stack, 'space-metrics', spaceMetricsTableProps)
 
   return {
+    allocationTable,
     storeTable,
     uploadTable,
     pieceTable,

upload-api/functions/ucan-invocation-router.js

@@ -23,6 +23,10 @@ import { createUcantoServer } from '../service.js'
 import { Config } from 'sst/node/config'
 import { CAR, Legacy, Codec } from '@ucanto/transport'
 import { Email } from '../email.js'
+import { createTasksStorage } from '../stores/tasks.js'
+import { createReceiptsStorage } from '../stores/receipts.js'
+import { createAllocationsStorage } from '../stores/allocations.js'
+import { createBlobsStorage, composeblobStoragesWithOrderedHas } from '../stores/blobs.js'
 import { useProvisionStore } from '../stores/provisions.js'
 import { useSubscriptionsStore } from '../stores/subscriptions.js'
 import { createDelegationsTable } from '../tables/delegations.js'
@@ -39,6 +43,7 @@ import { createSpaceDiffStore } from '@web3-storage/w3infra-billing/tables/space
 import { createSpaceSnapshotStore } from '@web3-storage/w3infra-billing/tables/space-snapshot.js'
 import { useUsageStore } from '../stores/usage.js'
 import { createStripeBillingProvider } from '../billing.js'
+import { createTasksScheduler } from '../scheduler.js'
 
 Sentry.AWSLambda.init({
   environment: process.env.SST_STAGE,
@@ -99,6 +104,7 @@ export async function ucanInvocationRouter(request) {
     storeTableName,
     storeBucketName,
     uploadTableName,
+    allocationTableName,
     consumerTableName,
     customerTableName,
     subscriptionTableName,
@@ -122,6 +128,7 @@ export async function ucanInvocationRouter(request) {
     aggregatorDid,
     dealTrackerDid,
     dealTrackerUrl,
+    uploadServiceURL,
     pieceOfferQueueUrl,
     filecoinSubmitQueueUrl,
     requirePaymentPlan,
@@ -144,6 +151,22 @@ export async function ucanInvocationRouter(request) {
   const { PRIVATE_KEY, STRIPE_SECRET_KEY } = Config
   const serviceSigner = getServiceSigner({ did: UPLOAD_API_DID, privateKey: PRIVATE_KEY })
 
+  const tasksStorage = createTasksStorage(AWS_REGION, invocationBucketName, workflowBucketName)
+  const receiptsStorage = createReceiptsStorage(AWS_REGION, taskBucketName, invocationBucketName, workflowBucketName)
+  const allocationsStorage = createAllocationsStorage(AWS_REGION, allocationTableName, {
+    endpoint: dbEndpoint,
+  })
+  const blobsStorage = composeblobStoragesWithOrderedHas(
+    createBlobsStorage(R2_REGION, carparkBucketName, {
+      endpoint: carparkBucketEndpoint,
+      credentials: {
+        accessKeyId: carparkBucketAccessKeyId,
+        secretAccessKey: carparkBucketSecretAccessKey,
+      }, 
+    }),
+    createBlobsStorage(AWS_REGION, storeBucketName),
+  )
+
   const invocationBucket = createInvocationStore(
     AWS_REGION,
     invocationBucketName
@@ -172,16 +195,29 @@ export async function ucanInvocationRouter(request) {
   const spaceSnapshotStore = createSpaceSnapshotStore({ region: AWS_REGION }, { tableName: spaceSnapshotTableName })
   const usageStorage = useUsageStore({ spaceDiffStore, spaceSnapshotStore })
 
-  const connection = getServiceConnection({
+  const dealTrackerConnection = getServiceConnection({
     did: dealTrackerDid,
     url: dealTrackerUrl
   })
+  const selfConnection = getServiceConnection({
+    did: serviceSigner.did(),
+    url: uploadServiceURL
+  })
+  const tasksScheduler = createTasksScheduler(() => selfConnection)
 
   const server = createUcantoServer(serviceSigner, {
     codec,
+    allocationsStorage,
+    blobsStorage,
+    tasksStorage,
+    receiptsStorage,
+    tasksScheduler,
+    getServiceConnection: () => selfConnection,
+    // TODO: to be deprecated with `store/*` protocol
     storeTable: createStoreTable(AWS_REGION, storeTableName, {
       endpoint: dbEndpoint,
     }),
+    // TODO: to be deprecated with `store/*` protocol
     carStoreBucket: composeCarStoresWithOrderedHas(
       createCarStore(AWS_REGION, storeBucketName),
       createCarStore(R2_REGION, carparkBucketName, {
@@ -192,6 +228,7 @@ export async function ucanInvocationRouter(request) {
         }, 
       }),
     ),
+    // TODO: to be deprecated with `store/*` protocol
     dudewhereBucket: createDudewhereStore(R2_REGION, R2_DUDEWHERE_BUCKET_NAME, {
       endpoint: R2_ENDPOINT,
       credentials: {
@@ -218,10 +255,10 @@ export async function ucanInvocationRouter(request) {
     pieceOfferQueue: createPieceOfferQueueClient({ region: AWS_REGION }, { queueUrl: pieceOfferQueueUrl }),
     filecoinSubmitQueue: createFilecoinSubmitQueueClient({ region: AWS_REGION }, { queueUrl: filecoinSubmitQueueUrl }),
     dealTrackerService: {
-      connection,
+      connection: dealTrackerConnection,
       invocationConfig: {
         issuer: serviceSigner,
-        audience: connection.id,
+        audience: dealTrackerConnection.id,
         with: serviceSigner.did()
       }
     },
@@ -316,6 +353,7 @@ function getLambdaEnv () {
     storeTableName: mustGetEnv('STORE_TABLE_NAME'),
     storeBucketName: mustGetEnv('STORE_BUCKET_NAME'),
     uploadTableName: mustGetEnv('UPLOAD_TABLE_NAME'),
+    allocationTableName: mustGetEnv('ALLOCATION_TABLE_NAME'),
     consumerTableName: mustGetEnv('CONSUMER_TABLE_NAME'),
     customerTableName: mustGetEnv('CUSTOMER_TABLE_NAME'),
     subscriptionTableName: mustGetEnv('SUBSCRIPTION_TABLE_NAME'),
@@ -339,6 +377,7 @@ function getLambdaEnv () {
     postmarkToken: mustGetEnv('POSTMARK_TOKEN'),
     providers: mustGetEnv('PROVIDERS'),
     accessServiceURL: mustGetEnv('ACCESS_SERVICE_URL'),
+    uploadServiceURL: mustGetEnv('UPLOAD_SERVICE_URL'),
     aggregatorDid: mustGetEnv('AGGREGATOR_DID'),
     requirePaymentPlan: (process.env.REQUIRE_PAYMENT_PLAN === 'true'),
     dealTrackerDid: mustGetEnv('DEAL_TRACKER_DID'),