feat: handle access/delegate invocations without error (#427)

Note:
* this is a superset of
#420
* that defines the `access/delegate` capability parsers that this PR
adds handlers for in access-api
  * my intention is to land that #420 before this

Motivation:
* part of #425

packages/access-api/migrations/0005_drop_delegations_audience_to_accounts_did_fk.sql

@@ -0,0 +1,30 @@
+-- Migration number: 0005 	 2023-02-09T23:48:40.469Z
+
+/*
+goal: remove the foreign key constraint on delegations.audience -> accounts.did.
+We want to be able to store delegations whose audience is not an account did.
+
+sqlite doesn't support `alter table drop constraint`.
+So here we will:
+* create delegations_new table without the constraint
+* insert all from delegations -> delegations_new
+* rename delegations_new -> delegations
+*/
+
+CREATE TABLE
+    IF NOT EXISTS delegations_new (
+        cid TEXT NOT NULL PRIMARY KEY,
+        bytes BLOB NOT NULL,
+        audience TEXT NOT NULL,
+        issuer TEXT NOT NULL,
+        expiration TEXT,
+        inserted_at TEXT NOT NULL DEFAULT (strftime ('%Y-%m-%dT%H:%M:%fZ', 'now')),
+        updated_at TEXT NOT NULL DEFAULT (strftime ('%Y-%m-%dT%H:%M:%fZ', 'now')),
+        UNIQUE (cid)
+    );
+
+INSERT INTO delegations_new (cid, bytes, audience, issuer, expiration, inserted_at, updated_at)
+SELECT cid, bytes, audience, issuer, expiration, inserted_at, updated_at FROM delegations;
+
+DROP TABLE delegations;
+ALTER TABLE delegations_new RENAME TO delegations;

packages/access-api/package.json

@@ -22,6 +22,7 @@
     "@ucanto/principal": "^4.2.3",
     "@ucanto/server": "^4.2.3",
     "@ucanto/transport": "^4.2.3",
+    "@ucanto/validator": "^4.2.3",
     "@web3-storage/access": "workspace:^",
     "@web3-storage/capabilities": "workspace:^",
     "@web3-storage/worker-utils": "0.4.3-dev",
@@ -31,6 +32,7 @@
     "preact": "^10.11.3",
     "preact-render-to-string": "^5.2.6",
     "qrcode": "^1.5.1",
+    "streaming-iterables": "^7.1.0",
     "toucan-js": "^2.7.0"
   },
   "devDependencies": {
@@ -85,7 +87,23 @@
       "WebSocketPair": "readonly"
     },
     "rules": {
-      "unicorn/prefer-number-properties": "off"
+      "unicorn/prefer-number-properties": "off",
+      "jsdoc/no-undefined-types": [
+        "error",
+        {
+          "definedTypes": [
+            "AsyncIterableIterator",
+            "Awaited",
+            "D1Database",
+            "FetchEvent",
+            "Iterable",
+            "IterableIterator",
+            "KVNamespace",
+            "PromiseLike",
+            "ResponseInit"
+          ]
+        }
+      ]
     }
   },
   "eslintIgnore": [

packages/access-api/src/bindings.d.ts

@@ -11,6 +11,7 @@ import { Validations } from './models/validations.js'
 import { loadConfig } from './config.js'
 import { ConnectionView, Signer as EdSigner } from '@ucanto/principal/ed25519'
 import { Accounts } from './models/accounts.js'
+import { DelegationsStorage as Delegations } from './types/delegations.js'
 
 export {}
 
@@ -59,6 +60,7 @@ export interface RouteContext {
     spaces: Spaces
     validations: Validations
     accounts: Accounts
+    delegations: Delegations
   }
   uploadApi: ConnectionView
 }

packages/access-api/src/models/delegations.js

@@ -0,0 +1,135 @@
+import * as Ucanto from '@ucanto/interface'
+import {
+  delegationsToBytes,
+  bytesToDelegations,
+} from '@web3-storage/access/encoding'
+
+/**
+ * @typedef {import('@web3-storage/access/src/types').DelegationTable} DelegationRow
+ * @typedef {Omit<DelegationRow, 'inserted_at'|'updated_at'|'expires_at'>} DelegationRowUpdate
+ */
+
+/**
+ * @typedef Tables
+ * @property {DelegationRow} delegations
+ */
+
+/**
+ * @typedef {import("../types/database").Database<Tables>} DelegationsDatabase
+ */
+
+/**
+ * DelegationsStorage that persists using SQL.
+ * * should work with cloudflare D1
+ */
+export class DbDelegationsStorage {
+  /** @type {DelegationsDatabase} */
+  #db
+
+  /**
+   * @param {DelegationsDatabase} db
+   */
+  constructor(db) {
+    this.#db = db
+    // eslint-disable-next-line no-void
+    void (
+      /** @type {import('../types/delegations').DelegationsStorage} */ (this)
+    )
+  }
+
+  async count() {
+    const { size } = await this.#db
+      .selectFrom('delegations')
+      .select((e) => e.fn.count('cid').as('size'))
+      .executeTakeFirstOrThrow()
+    return BigInt(size)
+  }
+
+  /**
+   * @param {import('../types/delegations').Query} query
+   */
+  async *find(query) {
+    for await (const row of await selectByAudience(this.#db, query.audience)) {
+      yield rowToDelegation(row)
+    }
+  }
+
+  /**
+   * store items
+   *
+   * @param  {Array<Ucanto.Delegation>} delegations
+   * @returns {Promise<void>}
+   */
+  async putMany(...delegations) {
+    if (delegations.length === 0) {
+      return
+    }
+    const values = delegations.map((d) => createDelegationRowUpdate(d))
+    await this.#db
+      .insertInto('delegations')
+      .values(values)
+      .onConflict((oc) => oc.column('cid').doNothing())
+      .executeTakeFirst()
+  }
+
+  /**
+   * iterate through all stored items
+   *
+   * @returns {AsyncIterableIterator<Ucanto.Delegation>}
+   */
+  async *[Symbol.asyncIterator]() {
+    if (!this.#db.canStream) {
+      throw Object.assign(
+        new Error(
+          `cannot create asyncIterator because the underlying database does not support streaming`
+        ),
+        { name: 'NotImplementedError' }
+      )
+    }
+    for await (const row of this.#db
+      .selectFrom('delegations')
+      .select(['bytes'])
+      .stream()) {
+      yield rowToDelegation(row)
+    }
+  }
+}
+
+/**
+ * @param {Pick<DelegationRow, 'bytes'>} row
+ * @returns {Ucanto.Delegation}
+ */
+function rowToDelegation(row) {
+  const delegations = bytesToDelegations(row.bytes)
+  if (delegations.length !== 1) {
+    throw new Error(
+      `unexpected number of delegations from bytes: ${delegations.length}`
+    )
+  }
+  return delegations[0]
+}
+
+/**
+ * @param {Ucanto.Delegation} d
+ * @returns {DelegationRowUpdate}
+ */
+function createDelegationRowUpdate(d) {
+  return {
+    cid: d.cid.toV1().toString(),
+    audience: d.audience.did(),
+    issuer: d.issuer.did(),
+    bytes: delegationsToBytes([d]),
+  }
+}
+
+/**
+ * @param {DelegationsDatabase} db
+ * @param {Ucanto.DID<'key'>} audience
+ */
+async function selectByAudience(db, audience) {
+  return await db
+    .selectFrom('delegations')
+    .selectAll()
+    .where('delegations.audience', '=', audience)
+    .execute()
+}

packages/access-api/src/models/spaces.js

@@ -21,7 +21,14 @@ export class Spaces {
   constructor(d1) {
     /** @type {GenericPlugin<SpaceRecord>} */
     const objectPlugin = new GenericPlugin({
-      metadata: (v) => JSON.parse(v),
+      metadata: (v) => {
+        // this will be `EMPTY` because it's the default value in the sql schema
+        // https://github.com/web3-storage/w3protocol/issues/447
+        if (v === 'EMPTY') {
+          return
+        }
+        return JSON.parse(v)
+      },
       inserted_at: (v) => new Date(v),
       updated_at: (v) => new Date(v),
     })

packages/access-api/src/service/access-claim.js

@@ -0,0 +1,54 @@
+import * as Server from '@ucanto/server'
+import { claim } from '@web3-storage/capabilities/access'
+import * as Ucanto from '@ucanto/interface'
+import * as validator from '@ucanto/validator'
+import * as delegationsResponse from '../utils/delegations-response.js'
+import { collect } from 'streaming-iterables'
+
+/**
+ * @typedef {import('@web3-storage/capabilities/types').AccessClaimSuccess} AccessClaimSuccess
+ * @typedef {import('@web3-storage/capabilities/types').AccessClaimFailure} AccessClaimFailure
+ */
+
+/**
+ * @callback AccessClaimHandler
+ * @param {Ucanto.Invocation<import('@web3-storage/capabilities/types').AccessClaim>} invocation
+ * @returns {Promise<Ucanto.Result<AccessClaimSuccess, AccessClaimFailure>>}
+ */
+
+/**
+ * @param {object} ctx
+ * @param {import('../types/delegations').DelegationsStorage} ctx.delegations
+ * @param {Pick<import('../bindings.js').RouteContext['config'], 'ENV'>} ctx.config
+ */
+export function accessClaimProvider(ctx) {
+  const handleClaimInvocation = createAccessClaimHandler(ctx)
+  return Server.provide(claim, async ({ invocation }) => {
+    // disable until hardened in test/staging
+    if (ctx.config.ENV === 'production') {
+      throw new Error(`acccess/claim invocation handling is not enabled`)
+    }
+    return handleClaimInvocation(invocation)
+  })
+}
+
+/**
+ * @param {object} options
+ * @param {import('../types/delegations').DelegationsStorage} options.delegations
+ * @returns {AccessClaimHandler}
+ */
+export function createAccessClaimHandler({ delegations }) {
+  /** @type {AccessClaimHandler} */
+  return async (invocation) => {
+    const claimedAudience = invocation.capabilities[0].with
+    if (validator.DID.match({ method: 'mailto' }).is(claimedAudience)) {
+      throw new Error(`did:mailto not supported`)
+    }
+    const claimed = await collect(
+      delegations.find({ audience: claimedAudience })
+    )
+    return {
+      delegations: delegationsResponse.encode(claimed),
+    }
+  }
+}