feat: add ucanto aggregation api service

package.json

@@ -12,7 +12,8 @@
     "deploy": "sst deploy",
     "remove": "sst remove",
     "console": "sst console",
-    "typecheck": "tsc --noEmit"
+    "typecheck": "tsc --noEmit",
+    "clean": "rm -rf node_modules pnpm-lock.yml packages/*/{pnpm-lock.yml,.next,out,coverage,.nyc_output,worker,dist,node_modules}"
   },
   "devDependencies": {
     "@sentry/serverless": "^7.52.1",
@@ -22,6 +23,7 @@
     "aws-cdk-lib": "2.72.1",
     "constructs": "10.1.156",
     "git-rev-sync": "^3.0.2",
+    "ts-node": "^10.9.1",
     "typescript": "^5.0.4",
     "@tsconfig/node16": "^1.0.3"
   },

packages/core/buckets/offer-store.ts

@@ -0,0 +1,47 @@
+import {
+  S3Client,
+  ServiceInputTypes,
+  PutObjectCommand,
+} from '@aws-sdk/client-s3'
+import pRetry from 'p-retry'
+
+import * as AggregateAPI from '@web3-storage/aggregate-api'
+
+export function createOfferStore(region: string, bucketName: string, options?: ServiceInputTypes) {
+  const s3 = new S3Client({
+    region,
+    ...(options || {}),
+  })
+  return useOfferStore(s3, bucketName)
+}
+
+export function useOfferStore(s3client: S3Client, bucketName: string) {
+  return {
+    put: async (commitmentProof: string, offers: AggregateAPI.Offer[]) => {
+      const putCmd = new PutObjectCommand({
+        Bucket: bucketName,
+        ContentType: 'application/json',
+        Key: `${getNextUtcDateName()} ${commitmentProof}`,
+        Body: JSON.stringify({
+          commitmentProof,
+          offers
+        })
+      })
+
+      await pRetry(() => s3client.send(putCmd))
+    }
+  } as AggregateAPI.OfferStore
+}
+
+export function getNextUtcDateName() {
+  const cDate = new Date()
+
+  // normalize date to multiple of 15 minutes
+  const currentMinute = cDate.getUTCMinutes()
+  const factor = Math.floor(currentMinute / 15) + 1
+  const additionalTime = ((factor * 15) - currentMinute) * 60000
+
+  const nDate = new Date(cDate.getTime() + additionalTime)
+
+  return `${nDate.getUTCFullYear()}-${nDate.getUTCMonth()}-${nDate.getUTCDay()} ${nDate.getUTCHours()}:${nDate.getUTCMinutes()}:00`
+}

packages/core/package.json

@@ -0,0 +1,37 @@
+{
+  "name": "@spade-proxy/core",
+  "version": "0.0.0",
+  "type": "module",
+  "scripts": {
+    "test": "ava --node-arguments='--experimental-fetch' --verbose --timeout=60s 'test/**.test.js'",
+    "typecheck": "tsc -noEmit"
+  },
+  "dependencies": {
+    "@aws-sdk/client-s3": "^3.338.0",
+    "@ipld/dag-ucan": "3.3.2",
+    "@ucanto/interface": "8.0.0",
+    "@ucanto/principal": "^8.0.0",
+    "@web3-storage/capabilities": "file:../../../w3up/packages/capabilities",
+    "@web3-storage/aggregate-api": "file:../../../w3up/packages/aggregate-api",
+    "@web3-storage/aggregate-client": "file:../../../w3up/packages/aggregate-client",
+    "p-retry": "^5.1.2"
+  },
+  "devDependencies": {
+    "@types/node": "^18.16.3",
+    "@web-std/blob": "^3.0.4",
+    "ava": "^4.3.3",
+    "nanoid": "^4.0.0",
+    "sst": "^2.8.3",
+    "testcontainers": "^8.13.0",
+    "typescript": "^5.0.4"
+  },
+  "ava": {
+    "extensions": {
+      "ts": "module",
+      "js": true
+    },
+    "nodeArguments": [
+      "--loader=ts-node/esm"
+    ]
+  }
+}

packages/core/service.ts

@@ -0,0 +1,31 @@
+import * as ed25519 from '@ucanto/principal/ed25519'
+import * as DID from '@ipld/dag-ucan/did'
+import { Signer } from '@ucanto/interface'
+
+import * as AggregateAPI from '@web3-storage/aggregate-api'
+
+export const createUcantoServer = (servicePrincipal: Signer, context: AggregateAPI.ServiceContext, errorReporter: AggregateAPI.ErrorReporter) =>
+AggregateAPI.createServer({
+  ...context,
+  id: servicePrincipal,
+  errorReporter
+})
+
+/**
+ * Given a config, return a ucanto Signer object representing the service
+ */
+ export function getServiceSigner(config: ServiceSignerCtx) {
+  const signer = ed25519.parse(config.PRIVATE_KEY)
+  if (config.SPADE_PROXY_DID) {
+    const did = DID.parse(config.SPADE_PROXY_DID).did()
+    return signer.withDID(did)
+  }
+  return signer
+}
+
+export type ServiceSignerCtx = {
+  // multiformats private key of primary signing key
+  PRIVATE_KEY: string
+  // public DID for the upload service (did:key:... derived from PRIVATE_KEY if not set)
+  SPADE_PROXY_DID: string
+}

packages/core/tables/aggregate-store.ts

@@ -0,0 +1,13 @@
+import * as AggregateAPI from '@web3-storage/aggregate-api'
+
+export function createAggregateStore() {
+  return useAggregateStore()
+}
+
+export function useAggregateStore() {
+  return {
+    get: async (commitmentProof: string) => {
+      throw new Error('not yet implemented')
+    }
+  } as AggregateAPI.AggregateStore
+}

packages/core/tables/arranged-offer-store.ts

@@ -0,0 +1,13 @@
+import * as AggregateAPI from '@web3-storage/aggregate-api'
+
+export function createArrangedOfferStore() {
+  return useArrangedOfferStore()
+}
+
+export function useArrangedOfferStore() {
+  return {
+    get: async (commitmentProof: string) => {
+      throw new Error('not yet implemented')
+    }
+  } as AggregateAPI.ArrangedOfferStore
+}

packages/core/test/helpers/context.js

@@ -0,0 +1,11 @@
+import anyTest from 'ava'
+
+/**
+ * @typedef {object} S3Context
+ * @property {import('@aws-sdk/client-s3').S3Client} s3
+ * 
+ * @typedef {import("ava").TestFn<S3Context>} Test
+ */
+
+// eslint-disable-next-line unicorn/prefer-export-from
+export const test = /** @type {Test} */ (anyTest)

packages/core/test/helpers/resources.js

@@ -0,0 +1,44 @@
+import { customAlphabet } from 'nanoid'
+import { GenericContainer as Container } from 'testcontainers'
+import { S3Client, CreateBucketCommand } from '@aws-sdk/client-s3'
+
+/**
+ * @param {object} [opts]
+ * @param {number} [opts.port]
+ * @param {string} [opts.region]
+ */
+export async function createS3(opts = {}) {
+  const region = opts.region || 'us-west-2'
+  const port = opts.port || 9000
+
+  const minio = await new Container('quay.io/minio/minio')
+    .withCmd(['server', '/data'])
+    .withExposedPorts(port)
+    .start()
+
+  const clientOpts = {
+    endpoint: `http://${minio.getHost()}:${minio.getMappedPort(port)}`,
+    forcePathStyle: true,
+    region,
+    credentials: {
+      accessKeyId: 'minioadmin',
+      secretAccessKey: 'minioadmin',
+    },
+  }
+
+  return {
+    client: new S3Client(clientOpts),
+    clientOpts,
+  }
+}
+
+/**
+ * @param {S3Client} s3
+ */
+export async function createBucket(s3) {
+  const id = customAlphabet('1234567890abcdefghijklmnopqrstuvwxyz', 10)
+  const Bucket = id()
+  await s3.send(new CreateBucketCommand({ Bucket }))
+  return Bucket
+}
+

packages/core/test/service.test.js

@@ -0,0 +1,80 @@
+/* eslint-disable no-loop-func */
+import { Aggregate } from '@web3-storage/aggregate-api/test'
+import { ed25519 } from '@ucanto/principal'
+
+import { test } from './helpers/context.js'
+import { createS3, createBucket } from './helpers/resources.js'
+
+import { useOfferStore } from '../buckets/offer-store.js'
+// TODO: Include official one
+// import { useAggregateStore } from '../tables/aggregate-store.js'
+import { useArrangedOfferStore } from '../tables/arranged-offer-store.js'
+
+test.before(async (t) => {
+  Object.assign(t.context, {
+    s3: (await createS3()).client,
+  })
+})
+
+for (const [title, unit] of Object.entries(Aggregate.test)) {
+  test(title, async (t) => {
+    const { s3 } = t.context
+    const bucketName = await createBucket(s3)
+
+    const offerStore = useOfferStore(s3, bucketName)
+    const arrangedOfferStore = useArrangedOfferStore()
+    const aggregateStore = getAggregateStore()
+
+    const signer = await ed25519.generate()
+    const id = signer.withDID('did:web:test.web3.storage')
+
+    await unit(
+      {
+        ok: (actual, message) => t.truthy(actual, message),
+        equal: (actual, expect, message) =>
+          t.is(actual, expect, message ? String(message) : undefined),
+        deepEqual: (actual, expect, message) =>
+          t.deepEqual(actual, expect, message ? String(message) : undefined),
+      },
+      {
+        id,
+        errorReporter: {
+          catch(error) {
+            t.fail(error.message)
+          },
+        },
+        offerStore,
+        aggregateStore,
+        aggregateStoreBackend: aggregateStore,
+        arrangedOfferStore
+      }
+    )
+  })
+}
+
+function getAggregateStore () {
+  /** @type {Map<string, unknown[]>} */
+  const items = new Map()
+
+  /** @type {import('@web3-storage/aggregate-api').AggregateStoreBackend & import('@web3-storage/aggregate-api').AggregateStore} */
+  const store = {
+    get: async (commitmentProof) => {
+      return Promise.resolve(items.get(commitmentProof))
+    },
+    put: async (commitmentProof, deal) => {
+      const dealEntries = items.get(commitmentProof)
+      let newEntries
+      if (dealEntries) {
+        newEntries = [...dealEntries, deal]
+        items.set(commitmentProof, newEntries)
+      } else {
+        newEntries = [deal]
+        items.set(commitmentProof, newEntries)
+      }
+
+      return Promise.resolve()
+    }
+  }
+
+  return store
+}

packages/core/tsconfig.json

@@ -0,0 +1,6 @@
+{
+  "extends": "@tsconfig/node16/tsconfig.json",
+  "compilerOptions": {
+    "module": "esnext"
+  }
+}

packages/functions/package.json

@@ -6,10 +6,15 @@
     "test": "sst bind vitest",
     "typecheck": "tsc -noEmit"
   },
+  "dependencies": {
+    "@spade-proxy/core": "*",
+    "@ucanto/server": "^8.0.0"
+  },
   "devDependencies": {
     "@sentry/serverless": "7.52.1",
     "@types/aws-lambda": "^8.10.115",
     "@types/node": "^18.16.3",
+    "@ucanto/core": "^8.0.0",
     "vitest": "^0.31.0",
     "sst": "^2.8.3",
     "typescript": "^5.0.4"