feat: offer arrange invocation receipt

packages/core/arrange-offers.ts

@@ -1,5 +1,8 @@
 import all from 'it-all'
 import * as AggregateAPI from '@web3-storage/aggregate-api'
+import * as Offer from '@web3-storage/capabilities/offer'
+import { Signer } from '@ucanto/interface'
+import { ConnectionView } from '@ucanto/interface'
 
 import { ArrangedOfferStore, ArrangedOffer } from './tables/arranged-offer-store'
 
@@ -30,4 +33,23 @@ export async function mutateOffersToArrange (arrangedOfferStore: ArrangedOfferSt
   return offersToArrange
 }
 
-// TODO: arrange offers
+export async function offerArrange(offer: ArrangedOffer, serviceSigner: Signer, conn: ConnectionView<any>) {
+  const { stat: status, commitmentProof } = offer
+
+  if (status !== 'accepted' && status !== 'rejected') {
+    // TODO: should this be a receipt of invocation with error?
+    throw new Error(`offer arrange for ${commitmentProof} cannot be fulfilled with given status: ${status}`)
+  }
+
+  const invocation = await Offer.arrange
+    .invoke({
+      issuer: serviceSigner,
+      audience: serviceSigner,
+      with: serviceSigner.did(),
+      nb: {
+        commitmentProof,
+      },
+    })
+
+  await invocation.execute(conn)
+}

packages/core/package.json

@@ -11,6 +11,7 @@
     "@aws-sdk/client-s3": "^3.338.0",
     "@aws-sdk/util-dynamodb": "^3.338.0",
     "@ipld/dag-ucan": "3.3.2",
+    "@ucanto/core": "8.0.0",
     "@ucanto/interface": "8.0.0",
     "@ucanto/principal": "^8.0.0",
     "@web3-storage/aggregate-api": "^0.0.0",

packages/core/test/arrange-offers.test.js

@@ -5,11 +5,13 @@ import {
   createDynamodDb,
   createTable,
 } from './helpers/resources.js'
-import { CID } from 'multiformats/cid'
 import { generateOffers, createAggregateStore } from './helpers/offers.js'
 
+import { CID } from 'multiformats/cid'
+import { ed25519 } from '@ucanto/principal'
+
 import { arrangedOfferTableProps } from '../tables/index.js'
-import { mutateOffersToArrange } from '../arrange-offers.js'
+import { mutateOffersToArrange, offerArrange } from '../arrange-offers.js'
 import { useOfferStore } from '../buckets/offer-store.js'
 import { useArrangedOfferStore } from '../tables/arranged-offer-store.js'
 
@@ -20,7 +22,7 @@ test.before(async (t) => {
   })
 })
 
-test('can arrange multiple offers', async t => {
+test('can mutate multiple offers to be arranged according to their state', async t => {
   const { dynamo, s3 } = t.context
   const bucketName = await createBucket(s3)
 
@@ -57,3 +59,43 @@ test('can arrange multiple offers', async t => {
   t.is(resWithSecondSegment.length, 1)
   t.is(resWithSecondSegment[0].commitmentProof.toString(), dataSegments[1].toString())
 })
+
+test('can create agent messages with receipt for arranged offer', async t => {
+  const signer = await ed25519.generate()
+  const commitmentProof = CID.parse('baga6ea4seaqcq4xx7rqx2lsrm6iky7qqk5jh7pbaj5bgdu22afhp4fodvccb6bq')
+
+  /** @type {import('../tables/arranged-offer-store.js').STAT[]} */
+  const stats = ['accepted', 'rejected']
+
+  for (const stat of stats) {
+    const offer = {
+      commitmentProof,
+      stat
+    }
+
+    const messageAcceptedReceipt = await offerArrange(offer, signer)
+
+    // console.log('message accepted', messageAcceptedReceipt.receipts)
+    t.is(messageAcceptedReceipt.receipts.size, 1)
+    const receiptCids = Array.from(messageAcceptedReceipt.receipts.keys())
+    const receipt = messageAcceptedReceipt.receipts.get(receiptCids[0])
+
+    // @ts-ignore unknown type
+    t.is(receipt?.out.ok?.status, stat)
+  }
+})
+
+test('fails to create agent message with receipt for non arranged offer', async t => {
+  const signer = await ed25519.generate()
+  const commitmentProof = CID.parse('baga6ea4seaqcq4xx7rqx2lsrm6iky7qqk5jh7pbaj5bgdu22afhp4fodvccb6bq')
+
+  /** @type {import('../tables/arranged-offer-store.js').STAT} */
+  const stat = 'queued'
+
+  const offer = {
+    commitmentProof,
+    stat
+  }
+
+  await t.throwsAsync(() => offerArrange(offer, signer))
+})

packages/functions/package.json

@@ -7,9 +7,13 @@
     "typecheck": "tsc -noEmit"
   },
   "dependencies": {
+    "@aws-sdk/util-dynamodb": "^3.338.0",
+    "@ipld/dag-ucan": "^3.3.2",
     "@spade-proxy/core": "*",
+    "@ucanto/client": "^8.0.0",
     "@ucanto/server": "^8.0.0",
-    "@ucanto/transport": "^8.0.0"
+    "@ucanto/transport": "^8.0.0",
+    "multiformats": "^11.0.2"
   },
   "devDependencies": {
     "@sentry/serverless": "7.52.1",

packages/functions/src/data/offer-arrange.ts

@@ -0,0 +1,74 @@
+import * as Sentry from '@sentry/serverless'
+import { Config } from 'sst/node/config'
+
+import { connect } from '@ucanto/client'
+import { CAR, HTTP } from '@ucanto/transport'
+import * as DID from '@ipld/dag-ucan/did'
+import { CID } from 'multiformats/cid'
+import { unmarshall } from '@aws-sdk/util-dynamodb'
+
+import { offerArrange } from '@spade-proxy/core/arrange-offers'
+import { getServiceSigner } from '@spade-proxy/core/service'
+
+import { mustGetEnv, parseDynamoDbEvent } from '../utils'
+
+Sentry.AWSLambda.init({
+  environment: process.env.SST_STAGE,
+  dsn: process.env.SENTRY_DSN,
+  tracesSampleRate: 1.0,
+})
+
+export async function main(event: import('aws-lambda').DynamoDBStreamEvent) {
+  const {
+    spadeProxyDid,
+    spadeProxyUrl
+  } = getLambdaEnv()
+
+  const records = parseDynamoDbEvent(event)
+  if (records.length > 1) {
+    throw new Error('Should only receive one offer to arrange')
+  }
+
+  // @ts-expect-error can't figure out type of new
+  const newRecord = unmarshall(records[0].new)
+
+  const { PRIVATE_KEY } = Config
+
+  const serviceSigner = getServiceSigner({ SPADE_PROXY_DID: spadeProxyDid, PRIVATE_KEY })
+  const serviceConnection = getServiceConnection({
+    DID: spadeProxyDid,
+    URL: spadeProxyUrl
+  })
+
+  // Create message with receipt from offer arrange
+  /** @type {import('@spade-proxy/core/tables/arranged-offer-store').ArrangedOffer} */
+  const offer = {
+    commitmentProof: CID.parse(newRecord.commitmentProof),
+    stat: newRecord.stat
+  }
+  await offerArrange(offer, serviceSigner, serviceConnection)
+}
+
+function getLambdaEnv () {
+  return {
+    spadeProxyDid: mustGetEnv('SPADE_PROXY_DID'),
+    spadeProxyUrl: mustGetEnv('SPADE_PROXY_URL')
+  }
+}
+
+export function getServiceConnection(config: { DID: string, URL: string}) {
+  const aggregationServicePrincipal = DID.parse(config.DID) // 'did:web:spade.web3.storage'
+  const aggregationServiceURL = new URL(config.URL) // 'https://spade-proxy.web3.storage'
+
+  const aggregationServiceConnection = connect({
+    id: aggregationServicePrincipal,
+    codec: CAR.outbound,
+    // @ts-expect-error
+    channel: HTTP.open({
+      url: aggregationServiceURL,
+      method: 'POST',
+    }),
+  })
+
+  return aggregationServiceConnection
+}

packages/functions/src/utils.ts

@@ -3,3 +3,10 @@ export function mustGetEnv (name: string) {
   if (!value) throw new Error(`missing ${name} environment variable`)
   return value
 }
+
+export function parseDynamoDbEvent (event: import('aws-lambda').DynamoDBStreamEvent) {
+  return event.Records.map(r => ({
+    new: r.dynamodb?.NewImage,
+    old: r.dynamodb?.OldImage
+  }))
+}

stacks/ApiStack.ts

@@ -1,6 +1,5 @@
 import {
   Api,
-  Config,
   StackContext,
   use
 } from 'sst/constructs';
@@ -14,7 +13,12 @@ import {
 } from './config';
 
 export function ApiStack({ app, stack }: StackContext) {
-  const { offerBucket, arrangedOfferTable } = use(DataStack)
+  const {
+    offerBucket,
+    arrangedOfferTable,
+    privateKey,
+    ucanLogBasicAuth
+  } = use(DataStack)
 
   // Setup app monitoring with Sentry
   setupSentry(app, stack)
@@ -23,14 +27,13 @@ export function ApiStack({ app, stack }: StackContext) {
   const customDomain = getCustomDomain(stack.stage, process.env.HOSTED_ZONE)
   const pkg = getApiPackageJson()
   const git = getGitInfo()
-  const privateKey = new Config.Secret(stack, 'PRIVATE_KEY')
-  const ucanLogBasicAuth = new Config.Secret(stack, 'UCAN_LOG_BASIC_AUTH')
 
   const api = new Api(stack, 'api', {
     customDomain,
     defaults: {
       function: {
         permissions: [
+          arrangedOfferTable,
           offerBucket
         ],
         environment: {

stacks/CronStack.ts

@@ -1,31 +1,40 @@
 import { Cron, StackContext, use } from 'sst/constructs';
 import { DataStack } from './DataStack';
 
+import {
+  getConstructName
+} from './config';
+
 export function CronStack({ stack }: StackContext) {
   const { offerBucket, arrangedOfferTable } = use(DataStack)
 
-  new Cron(stack, 'arrange-offers', {
+  const arrangeOffersCronName = getConstructName('arrange-offers-cron', stack.stage)
+  new Cron(stack, arrangeOffersCronName, {
+    // schedule: 'rate(1 minute)',
     schedule: 'rate(30 minutes)',
     job: {
       function: {
         handler: 'packages/functions/src/cron/trigger-offers-ready-to-arrange.main',
         environment: {
           ARRANGED_OFFER_TABLE_NAME: arrangedOfferTable.tableName,
-        }
+        },
+        permissions: [arrangedOfferTable]
       },
     }
   })
 
   // Write finalized files
-  const mergeOfferCron = new Cron(stack, 'merge-offer-cron', {
+  const mergeOfferCronName = getConstructName('merge-offer-cron', stack.stage)
+  const mergeOfferCron = new Cron(stack, mergeOfferCronName, {
     schedule: 'cron(0,15,30,45 * * * ? *)',
     job: {
       function: {
         handler: 'packages/functions/src/cron/merge-offers.main',
         environment: {
           OFFER_BUCKET_NAME: offerBucket.bucketName,
           ARRANGED_OFFER_TABLE_NAME: arrangedOfferTable.tableName,
-        }
+        },
+        permissions: [offerBucket, arrangedOfferTable]
       },
     }
   })