feat: persist ucan invocation car to s3 and replicate to r2

README.md

@@ -75,6 +75,10 @@ Bucket name to replicate written .idx files.
 
 Bucket name to replicate root CID to car CIDs mapping.
 
+#### `R2_UCAN_BUCKET_NAME`
+
+Bucket name to persist the CAR files of UCAN invocations handled by the service.
+
 #### `SENTRY_DSN`
 
 Data source name for Sentry application monitoring service.

api/buckets/ucan-store.js

@@ -0,0 +1,34 @@
+import { S3Client, PutObjectCommand } from '@aws-sdk/client-s3'
+
+/**
+ * Abstraction layer with Factory to perform operations on bucket storing
+ * handled UCANs
+ *
+ * @param {string} region
+ * @param {string} bucketName
+ * @param {import('@aws-sdk/client-s3').ServiceInputTypes} [options]
+ * @returns {import('../service/types').UcanBucket}
+ */
+export function createUcanStore (region, bucketName, options = {}) {
+  const s3client = new S3Client({
+    region,
+    ...options
+  })
+
+  return {
+    /**
+     * Put UCAN invocation CAR file into bucket
+     *
+     * @param {string} carCid
+     * @param {Uint8Array} bytes
+     */
+    put: async (carCid, bytes) => {
+      const putCmd = new PutObjectCommand({
+        Bucket: bucketName,
+        Key: carCid,
+        Body: bytes
+      })
+      await s3client.send(putCmd)
+    }
+  }
+}

api/functions/ucan-invocation-router.js

@@ -2,8 +2,10 @@ import { DID } from '@ucanto/core'
 import * as Sentry from '@sentry/serverless'
 
 import { createAccessClient } from '../access.js'
+import { persistUcanInvocation } from '../ucan-invocation.js'
 import { createCarStore } from '../buckets/car-store.js'
 import { createDudewhereStore } from '../buckets/dudewhere-store.js'
+import { createUcanStore } from '../buckets/ucan-store.js'
 import { createStoreTable } from '../tables/store.js'
 import { createUploadTable } from '../tables/upload.js'
 import { getServiceSigner } from '../config.js'
@@ -37,6 +39,7 @@ async function ucanInvocationRouter (request) {
     STORE_TABLE_NAME: storeTableName = '',
     STORE_BUCKET_NAME: storeBucketName = '',
     UPLOAD_TABLE_NAME: uploadTableName = '',
+    UCAN_BUCKET_NAME: ucanBucketName = '',
     // set for testing
     DYNAMO_DB_ENDPOINT: dbEndpoint,
     ACCESS_SERVICE_DID: accessServiceDID = '',
@@ -50,6 +53,8 @@ async function ucanInvocationRouter (request) {
   }
 
   const serviceSigner = getServiceSigner()
+  const ucanStoreBucket = createUcanStore(AWS_REGION, ucanBucketName)
+
   const server = await createUcantoServer(serviceSigner, {
     storeTable: createStoreTable(AWS_REGION, storeTableName, {
       endpoint: dbEndpoint
@@ -77,6 +82,9 @@ async function ucanInvocationRouter (request) {
     body: Buffer.from(request.body, 'base64'),
   })
 
+  // persist successful invocation handled
+  await persistUcanInvocation(request, ucanStoreBucket)
+
   return toLambdaSuccessResponse(response)
 }
 

api/service/index.js

@@ -39,4 +39,4 @@ export function createServiceRouter (context) {
   })
 
   return server
-}
+}

api/service/types.ts

@@ -54,6 +54,10 @@ export interface DudewhereBucket {
   put: (dataCid: string, carCid: string) => Promise<void>
 }
 
+export interface UcanBucket {
+  put: (carCid: string, bytes: Uint8Array) => Promise<void>
+}
+
 export interface StoreTable {
   exists: (space: DID, link: AnyLink) => Promise<boolean>
   insert: (item: StoreAddInput) => Promise<StoreAddOutput>

api/test/helpers/context.js

@@ -11,12 +11,20 @@ import anyTest from 'ava'
  * @property {string} accessServiceDID
  * @property {string} accessServiceURL
  * 
+ * @typedef {object} UcanInvocationContext
+ * @property {import('@aws-sdk/client-s3').S3Client} s3Client
+ * @property {import('@aws-sdk/client-s3').ServiceInputTypes} s3ClientOpts
+ * 
  * @typedef {import("ava").TestFn<Awaited<UcantoServerContext>>} TestStoreFn
+ * @typedef {import("ava").TestFn<Awaited<UcanInvocationContext>>} TestUcanInvocationFn
  * @typedef {import("ava").TestFn<Awaited<any>>} TestAnyFn
  */
 
 // eslint-disable-next-line unicorn/prefer-export-from
 export const testStore = /** @type {TestStoreFn} */ (anyTest)
 
+// eslint-disable-next-line unicorn/prefer-export-from
+export const testUcanInvocation = /** @type {TestUcanInvocationFn} */ (anyTest)
+
 // eslint-disable-next-line unicorn/prefer-export-from
 export const test  = /** @type {TestAnyFn} */ (anyTest)

api/test/service/ucan-invocation.test.js

@@ -0,0 +1,118 @@
+import { testUcanInvocation as test } from '../helpers/context.js'
+
+import { HeadObjectCommand } from '@aws-sdk/client-s3'
+import * as Signer from '@ucanto/principal/ed25519'
+import { CAR } from '@ucanto/transport'
+import * as UCAN from '@ipld/dag-ucan'
+
+import { createSpace } from '../helpers/ucanto.js'
+import { createS3, createBucket } from '../helpers/resources.js'
+
+import { createUcanStore } from '../../buckets/ucan-store.js'
+import { parseUcanInvocationRequest, persistUcanInvocation } from '../../ucan-invocation.js'
+
+test.before(async t => {
+  const { client: s3Client, clientOpts: s3ClientOpts } = await createS3({ port: 9000 })
+
+  t.context.s3Client = s3Client
+  t.context.s3ClientOpts = s3ClientOpts
+})
+
+test('parses ucan invocation request', async t => {
+  const uploadService = await Signer.generate()
+  const alice = await Signer.generate()
+  const { proof, spaceDid } = await createSpace(alice)
+
+  const data = new Uint8Array([11, 22, 34, 44, 55])
+  const link = await CAR.codec.link(data)
+  const nb = { link, size: data.byteLength }
+  const can = 'store/add'
+
+  const request = await CAR.encode([
+    {
+      issuer: alice,
+      audience: uploadService,
+      capabilities: [{
+        can,
+        with: spaceDid,
+        nb
+      }],
+      proofs: [proof],
+    }
+  ])
+
+  // @ts-expect-error different type interface in AWS expected request
+  const ucanInvocationObject = await parseUcanInvocationRequest(request)
+
+  const requestCar = await CAR.codec.decode(request.body)
+  const requestCarRootCid = requestCar.roots[0].cid
+
+  t.is(ucanInvocationObject.carCid, requestCarRootCid.toString())
+  t.truthy(ucanInvocationObject.bytes)
+
+  // Decode and validate bytes
+  const ucanCar = await CAR.codec.decode(ucanInvocationObject.bytes)
+  // @ts-expect-error UCAN.View<UCAN.Capabilities> inferred as UCAN.View<unknown>
+  const ucan = UCAN.decode(ucanCar.roots[0].bytes)
+
+  t.is(ucan.iss.did(), alice.did())
+  t.is(ucan.aud.did(), uploadService.did())
+  t.deepEqual(ucan.prf, [proof.root.cid])  
+  t.is(ucan.att.length, 1)
+  t.like(ucan.att[0], {
+    nb,
+    can,
+    with: spaceDid,
+  })
+})
+
+test('persists ucan invocation CAR file', async t => {
+  const { bucketName } = await prepareResources(t.context.s3Client)
+  const ucanStore = createUcanStore('us-west-2', bucketName, t.context.s3ClientOpts)
+
+  const uploadService = await Signer.generate()
+  const alice = await Signer.generate()
+  const { proof, spaceDid } = await createSpace(alice)
+
+  const data = new Uint8Array([11, 22, 34, 44, 55])
+  const link = await CAR.codec.link(data)
+  const nb = { link, size: data.byteLength }
+  const can = 'store/add'
+
+  const request = await CAR.encode([
+    {
+      issuer: alice,
+      audience: uploadService,
+      capabilities: [{
+        can,
+        with: spaceDid,
+        nb
+      }],
+      proofs: [proof],
+    }
+  ])
+
+  // @ts-expect-error different type interface in AWS expected request
+  await persistUcanInvocation(request, ucanStore)
+
+  const requestCar = await CAR.codec.decode(request.body)
+  const requestCarRootCid = requestCar.roots[0].cid
+
+  const cmd = new HeadObjectCommand({
+    Key: requestCarRootCid.toString(),
+    Bucket: bucketName,
+  })
+  const s3Response = await t.context.s3Client.send(cmd)
+  t.is(s3Response.$metadata.httpStatusCode, 200)
+})
+
+/**
+ * @param {import("@aws-sdk/client-s3").S3Client} s3Client
+ */
+async function prepareResources (s3Client) {
+  const bucketName = await createBucket(s3Client)
+
+  return {
+    bucketName
+  }
+}

api/ucan-invocation.js

@@ -0,0 +1,31 @@
+import * as CAR from '@ucanto/transport/car'
+
+/**
+ * Persist successful UCAN invocations handled by the router.
+ *
+ * @param {import('aws-lambda').APIGatewayProxyEventV2} request
+ * @param {import('./service/types').UcanBucket} ucanStore 
+ */
+export async function persistUcanInvocation (request, ucanStore) {
+  const { carCid, bytes } = await parseUcanInvocationRequest(request)
+
+  await ucanStore.put(carCid, bytes)
+}
+
+/**
+ * @param {import('aws-lambda').APIGatewayProxyEventV2} request
+ */
+export async function parseUcanInvocationRequest (request) {
+  if (!request.body) {
+    throw new Error('service requests are required to have body')
+  }
+
+  const bytes = Buffer.from(request.body, 'base64')
+  const car = await CAR.codec.decode(bytes)
+  const carCid = car.roots[0].cid.toString()
+
+  return {
+    bytes,
+    carCid
+  }
+}