feat: w3 aggregate protocol client and api implementation (#787)

This PR is a first iteration for w3 aggregate client and API
implementations. It follows same patterns we identified while working on
`upload-api` and `upload-client` and is agnostic from the deployment
service. `aggregate-api` exposes a lib and a set of tests that
implementers with infra should follow, as well as the needed interfaces
that must be fulfilled. It also includes `fx` usage for
`aggregate/offer` return value

It follows spec
https://github.com/web3-storage/specs/blob/feat/filecoin-spec/w3-aggregation.md

There are two interfaces defined in this service:
- `offerStore` - responsible for receiving offers to be queued.
Implementer receives `commD` of aggregate together with Offer content.
These will be used to give SPs. Note that implementer should keep track
of the offer's `commD` so that it can query over time when a deal is
done (or failures), so that the `fx` invocation can be executed and a
receipt generated
- `aggregateStore` - store of aggregates already places with SPs
(implementor like `spade-proxy` will have this store hooked with spade
DB)

New detailed issues for above will be created before this PR gets
merged.

Part of  storacha/w3filecoin-infra#19

Closes #772 and #773

packages/access-api/package.json

@@ -95,6 +95,7 @@
     },
     "rules": {
       "unicorn/prefer-number-properties": "off",
+      "unicorn/expiring-todo-comments": "off",
       "@typescript-eslint/ban-types": "off",
       "unicorn/no-null": "off",
       "jsdoc/no-undefined-types": [

packages/access-client/package.json

@@ -112,6 +112,7 @@
     "rules": {
       "unicorn/prefer-number-properties": "off",
       "unicorn/prefer-export-from": "off",
+      "unicorn/expiring-todo-comments": "off",
       "unicorn/no-array-reduce": "off",
       "unicorn/explicit-length-check": "off",
       "jsdoc/no-undefined-types": [

packages/aggregate-api/package.json

@@ -0,0 +1,113 @@
+{
+  "name": "@web3-storage/aggregate-api",
+  "version": "0.0.0",
+  "type": "module",
+  "main": "./src/lib.js",
+  "files": [
+    "src",
+    "test",
+    "dist/**/*.d.ts",
+    "dist/**/*.d.ts.map"
+  ],
+  "typesVersions": {
+    "*": {
+      "src/lib.js": [
+        "dist/src/lib.d.ts"
+      ],
+      "aggregate": [
+        "dist/src/aggregate.d.ts"
+      ],
+      "offer": [
+        "dist/src/offer.d.ts"
+      ],
+      "types": [
+        "dist/src/types.d.ts"
+      ],
+      "test": [
+        "dist/test/lib.d.ts"
+      ]
+    }
+  },
+  "exports": {
+    ".": {
+      "types": "./dist/src/lib.d.ts",
+      "import": "./src/lib.js"
+    },
+    "./types": {
+      "types": "./dist/src/types.d.ts",
+      "import": "./src/types.js"
+    },
+    "./aggregate": {
+      "types": "./dist/src/aggregate.d.ts",
+      "import": "./src/aggregate.js"
+    },
+    "./offer": {
+      "types": "./dist/src/offer.d.ts",
+      "import": "./src/offer.js"
+    },
+    "./test": {
+      "types": "./dist/test/lib.d.ts",
+      "import": "./test/lib.js"
+    }
+  },
+  "scripts": {
+    "build": "tsc --build",
+    "check": "tsc --build",
+    "lint": "tsc --build",
+    "test": "mocha --bail --timeout 10s -n no-warnings -n experimental-vm-modules -n experimental-fetch test/**/*.spec.js",
+    "test-watch": "pnpm build && mocha --bail --timeout 10s --watch --parallel -n no-warnings -n experimental-vm-modules -n experimental-fetch --watch-files src,test"
+  },
+  "dependencies": {
+    "@ucanto/client": "^8.0.0",
+    "@ucanto/core": "^8.0.0",
+    "@ucanto/interface": "^8.0.0",
+    "@ucanto/server": "^8.0.0",
+    "@ucanto/transport": "^8.0.0",
+    "@web3-storage/capabilities": "workspace:^"
+  },
+  "devDependencies": {
+    "@ipld/car": "^5.1.1",
+    "@types/mocha": "^10.0.1",
+    "@ucanto/principal": "^8.0.0",
+    "@web-std/blob": "^3.0.4",
+    "@web3-storage/aggregate-client": "workspace:^",
+    "hd-scripts": "^4.1.0",
+    "mocha": "^10.2.0",
+    "multiformats": "^11.0.2"
+  },
+  "eslintConfig": {
+    "extends": [
+      "./node_modules/hd-scripts/eslint/index.js"
+    ],
+    "parserOptions": {
+      "project": "./tsconfig.json"
+    },
+    "rules": {
+      "unicorn/expiring-todo-comments": "off"
+    },
+    "env": {
+      "mocha": true
+    },
+    "ignorePatterns": [
+      "dist",
+      "coverage"
+    ]
+  },
+  "depcheck": {
+    "specials": [
+      "bin"
+    ],
+    "ignorePatterns": [
+      "dist"
+    ],
+    "ignores": [
+      "dist",
+      "@types/*",
+      "hd-scripts",
+      "eslint-config-prettier"
+    ]
+  },
+  "engines": {
+    "node": ">=16.15"
+  }
+}

packages/aggregate-api/src/aggregate.js

@@ -0,0 +1,13 @@
+import { provide as aggregateOfferProvider } from './aggregate/offer.js'
+import { provide as aggregateGetProvider } from './aggregate/get.js'
+import * as API from './types.js'
+
+/**
+ * @param {API.AggregateServiceContext} context
+ */
+export function createService(context) {
+  return {
+    offer: aggregateOfferProvider(context),
+    get: aggregateGetProvider(context),
+  }
+}

packages/aggregate-api/src/aggregate/get.js

@@ -0,0 +1,38 @@
+import * as Server from '@ucanto/server'
+import * as Aggregate from '@web3-storage/capabilities/aggregate'
+import * as API from '../types.js'
+
+/**
+ * @param {API.AggregateServiceContext} context
+ */
+export const provide = (context) =>
+  Server.provide(Aggregate.get, (input) => claim(input, context))
+
+/**
+ * @param {API.Input<Aggregate.get>} input
+ * @param {API.AggregateServiceContext} context
+ * @returns {Promise<API.UcantoInterface.Result<API.AggregateGetSuccess, API.AggregateGetFailure>>}
+ */
+export const claim = async ({ capability }, { aggregateStore }) => {
+  const commitmentProof = capability.nb.commitmentProof
+
+  const aggregateArrangedResult = await aggregateStore.get(commitmentProof)
+  if (!aggregateArrangedResult) {
+    return {
+      error: new AggregateNotFound(
+        `aggregate not found for commitment proof: ${commitmentProof}`
+      ),
+    }
+  }
+  return {
+    ok: {
+      deals: aggregateArrangedResult,
+    },
+  }
+}
+
+class AggregateNotFound extends Server.Failure {
+  get name() {
+    return /** @type {const} */ ('AggregateNotFound')
+  }
+}

packages/aggregate-api/src/aggregate/offer.js

@@ -0,0 +1,133 @@
+import * as Server from '@ucanto/server'
+import { CBOR } from '@ucanto/core'
+import * as Aggregate from '@web3-storage/capabilities/aggregate'
+import * as Offer from '@web3-storage/capabilities/offer'
+import * as API from '../types.js'
+
+export const MIN_SIZE = 1 + 127 * (1 << 27)
+export const MAX_SIZE = 127 * (1 << 28)
+
+/**
+ * @param {API.AggregateServiceContext} context
+ */
+export const provide = (context) =>
+  Server.provideAdvanced({
+    capability: Aggregate.offer,
+    handler: (input) => claim(input, context),
+  })
+
+/**
+ * @param {API.Input<Aggregate.offer>} input
+ * @param {API.AggregateServiceContext} context
+ * @returns {Promise<API.UcantoInterface.Result<API.AggregateOfferSuccess, API.AggregateOfferFailure> | API.UcantoInterface.JoinBuilder<API.AggregateOfferSuccess>>}
+ */
+export const claim = async (
+  { capability, invocation, context },
+  { offerStore }
+) => {
+  // Get offer block
+  const offerCid = capability.nb.offer
+  const commitmentProof = capability.nb.commitmentProof
+  const offers = getOfferBlock(offerCid, invocation)
+
+  if (!offers) {
+    return {
+      error: new AggregateOfferBlockNotFoundError(
+        `missing offer block in invocation: ${offerCid.toString()}`
+      ),
+    }
+  }
+
+  // Validate offer content
+  const size = offers.reduce((accum, offer) => accum + offer.size, 0)
+  if (size < MIN_SIZE) {
+    return {
+      error: new AggregateOfferInvalidSizeError(
+        `offer under size, offered: ${size}, minimum: ${MIN_SIZE}`
+      ),
+    }
+  } else if (size > MAX_SIZE) {
+    return {
+      error: new AggregateOfferInvalidSizeError(
+        `offer over size, offered: ${size}, maximum: ${MAX_SIZE}`
+      ),
+    }
+  } else if (size !== capability.nb.size) {
+    return {
+      error: new AggregateOfferInvalidSizeError(
+        `offer size mismatch, specified: ${capability.nb.size}, actual: ${size}`
+      ),
+    }
+  }
+
+  // Validate URLs in offers src
+  for (const offer of offers.values()) {
+    for (const u of offer.src) {
+      try {
+        new URL(u)
+      } catch {
+        return {
+          error: new AggregateOfferInvalidUrlError(
+            `offer has invalid URL: ${u}`
+          ),
+        }
+      }
+    }
+  }
+
+  // TODO: Validate commP
+
+  // Create effect for receipt
+  const fx = await Offer.arrange
+    .invoke({
+      issuer: context.id,
+      audience: context.id,
+      with: context.id.did(),
+      nb: {
+        commitmentProof,
+      },
+    })
+    .delegate()
+
+  // Write offer to store
+  await offerStore.queue({ commitmentProof, offers })
+
+  return Server.ok({
+    status: 'queued',
+  }).join(fx.link())
+}
+
+/**
+ * @param {Server.API.Link<unknown, number, number, 0 | 1>} offerCid
+ * @param {Server.API.Invocation<Server.API.Capability<"aggregate/offer", `did:${string}:${string}` & `did:${string}` & Server.API.Phantom<{ protocol: "did:"; }> & `${string}:${string}` & Server.API.Phantom<{ protocol: `${string}:`; }>, Pick<{ offer: Server.API.Link<unknown, number, number, 0 | 1>; commitmentProof: Server.API.Link<unknown, number, number, 0 | 1>; size: number & Server.API.Phantom<{ typeof: "integer"; }>; }, "offer" | "commitmentProof" | "size"> & Partial<Pick<{ offer: Server.API.Link<unknown, number, number, 0 | 1>; commitmentProof: Server.API.Link<unknown, number, number, 0 | 1>; size: number & Server.API.Phantom<{ typeof: "integer"; }>; }, never>>>>} invocation
+ */
+function getOfferBlock(offerCid, invocation) {
+  for (const block of invocation.iterateIPLDBlocks()) {
+    if (block.cid.equals(offerCid)) {
+      const decoded =
+        /** @type {import('@web3-storage/aggregate-client/types').Offer[]} */ (
+          CBOR.decode(block.bytes)
+        )
+      return decoded
+      // TODO: Validate with schema
+    }
+  }
+}
+
+class AggregateOfferInvalidUrlError extends Server.Failure {
+  get name() {
+    return /** @type {const} */ ('AggregateOfferInvalidUrl')
+  }
+}
+
+class AggregateOfferInvalidSizeError extends Server.Failure {
+  get name() {
+    return /** @type {const} */ ('AggregateOfferInvalidSize')
+  }
+}
+
+class AggregateOfferBlockNotFoundError extends Server.Failure {
+  get name() {
+    return /** @type {const} */ ('AggregateOfferBlockNotFound')
+  }
+}

packages/aggregate-api/src/lib.js

@@ -0,0 +1,46 @@
+import * as Server from '@ucanto/server'
+import * as Client from '@ucanto/client'
+import * as Types from './types.js'
+import * as CAR from '@ucanto/transport/car'
+import { createService as createAggregateService } from './aggregate.js'
+import { createService as createOfferService } from './offer.js'
+export * from './types.js'
+
+/**
+ * @param {Types.UcantoServerContext} options
+ */
+export const createServer = ({ id, codec = CAR.inbound, ...context }) =>
+  Server.create({
+    id,
+    codec: CAR.inbound,
+    service: createService(context),
+    catch: (error) => context.errorReporter.catch(error),
+  })
+
+/**
+ * @param {Types.ServiceContext} context
+ * @returns {Types.Service}
+ */
+export const createService = (context) => ({
+  aggregate: createAggregateService(context),
+  offer: createOfferService(context),
+})
+
+/**
+ * @param {object} options
+ * @param {Types.UcantoInterface.Principal} options.id
+ * @param {Types.UcantoInterface.Transport.Channel<Types.Service>} options.channel
+ * @param {Types.UcantoInterface.OutboundCodec} [options.codec]
+ */
+export const connect = ({ id, channel, codec = CAR.outbound }) =>
+  Client.connect({
+    id,
+    channel,
+    codec,
+  })
+
+export {
+  createService as createUploadService,
+  createServer as createUploadServer,
+  connect as createUploadClient,
+}

packages/aggregate-api/src/offer.js

@@ -0,0 +1,11 @@
+import { provide as offerArrangeProvider } from './offer/arrange.js'
+import * as API from './types.js'
+
+/**
+ * @param {API.OfferServiceContext} context
+ */
+export function createService(context) {
+  return {
+    arrange: offerArrangeProvider(context),
+  }
+}