feat: move access-api delegation bytes out of d1 and into r2

package.json

@@ -22,8 +22,7 @@
     "lint-staged": "^13.2.0",
     "prettier": "2.8.3",
     "typedoc-plugin-markdown": "^3.14.0",
-    "typescript": "4.9.5",
-    "wrangler": "^2.12.3"
+    "typescript": "4.9.5"
   },
   "prettier": {
     "trailingComma": "es5",

packages/access-api/migrations/0007_add_delegations_v3.sql

@@ -0,0 +1,19 @@
+-- Migration number: 0007 	 2023-03-20T23:48:40.469Z
+
+/*
+goal: add a new table to store delegations in
+which doesn't have a 'bytes' column.
+
+context: we're going to start storing bytes outside of the database (e.g. in R2)
+*/
+
+CREATE TABLE
+    IF NOT EXISTS delegations_v3 (
+        cid TEXT NOT NULL PRIMARY KEY,
+        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)
+    );

packages/access-api/package.json

@@ -38,9 +38,10 @@
     "toucan-js": "^2.7.0"
   },
   "devDependencies": {
-    "@cloudflare/workers-types": "^3.18.0",
+    "@cloudflare/workers-types": "^3.19.0",
     "@databases/split-sql-query": "^1.0.3",
     "@databases/sql": "^3.2.0",
+    "@miniflare/r2": "^2.12.1",
     "@sentry/cli": "2.7.0",
     "@types/assert": "^1.5.6",
     "@types/git-rev-sync": "^2.0.0",
@@ -55,14 +56,14 @@
     "git-rev-sync": "^3.0.2",
     "hd-scripts": "^4.0.0",
     "is-subset": "^0.1.1",
-    "miniflare": "^2.11.0",
+    "miniflare": "^2.12.1",
     "mocha": "^10.2.0",
     "p-wait-for": "^5.0.0",
     "process": "^0.11.10",
     "readable-stream": "^4.2.0",
     "sade": "^1.8.1",
     "typescript": "4.9.5",
-    "wrangler": "^2.8.0"
+    "wrangler": "^2.13.0"
   },
   "eslintConfig": {
     "extends": [

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

@@ -1,9 +1,4 @@
 import type { Logging } from '@web3-storage/worker-utils/logging'
-import type {
-  AccountTable,
-  DelegationTable,
-  SpaceTable,
-} from '@web3-storage/access/types'
 import type { Handler as _Handler } from '@web3-storage/worker-utils/router'
 import { Spaces } from './models/spaces.js'
 import { Validations } from './models/validations.js'
@@ -108,9 +103,3 @@ export interface ModuleWorker {
 export interface D1ErrorRaw extends Error {
   cause: Error & { code: string }
 }
-
-export interface D1Schema {
-  spaces: SpaceTable
-  accounts: AccountTable
-  delegations: DelegationTable
-}

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

@@ -1,35 +1,29 @@
-// eslint-disable-next-line no-unused-vars
 import * as Ucanto from '@ucanto/interface'
 import { Kysely } from 'kysely'
 import { D1Dialect } from 'kysely-d1'
 import { GenericPlugin } from '../utils/d1.js'
 
-/**
- * @typedef {import('@web3-storage/access/src/types.js').DelegationRecord} DelegationRecord
- */
-
 /**
  * Accounts
  */
 export class Accounts {
   /**
-   *
    * @param {D1Database} d1
    */
   constructor(d1) {
-    /** @type {GenericPlugin<DelegationRecord>} */
     const objectPlugin = new GenericPlugin({
       // eslint-disable-next-line unicorn/no-null
       expires_at: (v) => (typeof v === 'string' ? new Date(v) : null),
       inserted_at: (v) => new Date(v),
       updated_at: (v) => new Date(v),
     })
-    this.d1 = /** @type {Kysely<import('../bindings').D1Schema>} */ (
-      new Kysely({
-        dialect: new D1Dialect({ database: d1 }),
-        plugins: [objectPlugin],
-      })
-    )
+    this.d1 =
+      /** @type {Kysely<{ accounts: import('@web3-storage/access/src/types.js').AccountTable }>} */ (
+        new Kysely({
+          dialect: new D1Dialect({ database: d1 }),
+          plugins: [objectPlugin],
+        })
+      )
   }
 
   /**

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

@@ -5,19 +5,24 @@ import {
 } from '@web3-storage/access/encoding'
 
 /**
- * @typedef {import('@web3-storage/access/src/types').DelegationTable} DelegationRow
+ * @typedef {import('../types/access-api-cf-db').R2Bucket} R2Bucket
+ */
+
+/**
+ * @template {import('../types/access-api-cf-db').DelegationsV2Table | import('../types/access-api-cf-db').DelegationsV3Table} DelegationRow
  * @typedef {Omit<DelegationRow, 'inserted_at'|'updated_at'|'expires_at'>} DelegationRowUpdate
  */
 
 /**
- * @typedef Tables
- * @property {DelegationRow} delegations_v2
+ * @typedef V2Tables
+ * @property {import('../types/access-api-cf-db').DelegationsV2Table} delegations_v2
  */
 
 /**
- * @typedef {import("../types/database").Database<Tables>} DelegationsDatabase
+ * @typedef {import("../types/database").Database<V2Tables>} DelegationsDatabase
  */
 
+// @todo remove reference to v2
 export const delegationsTable = /** @type {const} */ ('delegations_v2')
 
 /**
@@ -43,11 +48,7 @@ export class DbDelegationsStorage {
   }
 
   async count() {
-    const { size } = await this.#db
-      .selectFrom(this.#tables.delegations)
-      .select((e) => e.fn.count('cid').as('size'))
-      .executeTakeFirstOrThrow()
-    return BigInt(size)
+    return count(this.#db, this.#tables.delegations)
   }
 
   /**
@@ -116,14 +117,18 @@ class UnexpectedDelegation extends Error {
 }
 
 /**
- * @param {Pick<DelegationRow, 'bytes'>} row
+ * @param {Pick<import('../types/access-api-cf-db').DelegationsV2Table, 'bytes'>} row
  * @returns {Ucanto.Delegation}
  */
 function rowToDelegation(row) {
   /** @type {Ucanto.Delegation[]} */
   let delegations = []
+  // kysely/sqlite/d1/various-miniflare-versions sometimes only give an Array here.
+  // (GenericPlugin tries to ensure Uint8Array, but it can't reliably detect when to cast Array -> Uint8Array)
+  const rowBytes =
+    row.bytes instanceof Uint8Array ? row.bytes : new Uint8Array(row.bytes)
   try {
-    delegations = bytesToDelegations(row.bytes)
+    delegations = bytesToDelegations(rowBytes)
   } catch (error) {
     if (
       typeof error === 'object' &&
@@ -152,7 +157,7 @@ function rowToDelegation(row) {
 
 /**
  * @param {Ucanto.Delegation} d
- * @returns {DelegationRowUpdate}
+ * @returns {DelegationRowUpdate<import('../types/access-api-cf-db').DelegationsV2Table>}
  */
 export function createDelegationRowUpdate(d) {
   return {
@@ -163,6 +168,17 @@ export function createDelegationRowUpdate(d) {
   }
 }
 
+/**
+ * @param {Ucanto.Delegation} d
+ */
+export function createDelegationRowUpdateV3(d) {
+  return {
+    cid: d.cid.toV1().toString(),
+    audience: d.audience.did(),
+    issuer: d.issuer.did(),
+  }
+}
+
 /**
  * @param {Array<number> | Buffer | unknown} sqlValue - value from kysely 'bytes' table - in node it could be a Buffer. In cloudflare it might be an Array
  * @returns {ArrayBuffer|undefined} - undefined if unable to convert
@@ -179,3 +195,160 @@ export function delegationsTableBytesToArrayBuffer(sqlValue) {
     return Uint8Array.from(sqlValue)
   }
 }
+
+/**
+ * @typedef {`delegations_v3`} DelegationsTableWithoutBytesName
+ */
+
+/** @type {DelegationsTableWithoutBytesName} */
+export const delegationsV3Table = `delegations_v3`
+
+/**
+ * @typedef {import('../types/access-api-cf-db').AccessApiD1TablesV2} AccessApiD1TablesV2
+ * @typedef {import('../types/access-api-cf-db').AccessApiD1TablesV3} AccessApiD1TablesV3
+ */
+
+export class DbDelegationsStorageWithR2 {
+  // @todo abstract away R2 specifics into DagStore: ~AsyncMap<CID, Ucanto.Delegation>
+  /** @type {R2Bucket} */
+  #dags
+  /** @type {import('../types/database').Database<AccessApiD1TablesV3>} */
+  #db
+  /** @type {keyof AccessApiD1TablesV3} */
+  #delegationsTableName = delegationsV3Table
+
+  /**
+   * @param {import('../types/database').Database<AccessApiD1TablesV3>} db
+   * @param {R2Bucket} dags
+   * @param {keyof AccessApiD1TablesV3} delegationsTableName
+   */
+  // eslint-disable-next-line no-useless-constructor
+  constructor(db, dags, delegationsTableName = delegationsV3Table) {
+    this.#db = db
+    this.#delegationsTableName = delegationsTableName
+    this.#dags = dags
+  }
+
+  /**
+   * store items
+   *
+   * @param  {Array<Ucanto.Delegation>} delegations
+   * @returns {Promise<void>}
+   */
+  async putMany(...delegations) {
+    if (delegations.length === 0) {
+      return
+    }
+    await writeDelegations(this.#dags, delegations)
+    const values = delegations.map((d) => createDelegationRowUpdateV3(d))
+    await this.#db
+      .insertInto(this.#delegationsTableName)
+      .values(values)
+      .onConflict((oc) => oc.column('cid').doNothing())
+      .executeTakeFirst()
+  }
+
+  /** @returns {Promise<bigint>} */
+  async count() {
+    return count(this.#db, this.#delegationsTableName)
+  }
+
+  /**
+   * @param {import('../types/delegations').Query} query
+   */
+  async *find(query) {
+    const { audience } = query
+    const delegations = this.#delegationsTableName
+    const selection = await this.#db
+      .selectFrom(delegations)
+      .selectAll()
+      .where(`${delegations}.audience`, '=', audience)
+      .execute()
+    for await (const row of selection) {
+      yield this.#rowToDelegation(row)
+    }
+  }
+
+  /**
+   * @param {Pick<import('../types/access-api-cf-db').DelegationsV3Table, 'cid'>} row
+   * @param {R2Bucket} dags
+   * @returns {Promise<Ucanto.Delegation>}
+   */
+  async #rowToDelegation(row, dags = this.#dags) {
+    const { cid } = row
+    const carBytesR2 = await dags.get(cid.toString())
+    if (!carBytesR2) {
+      throw new Error(`failed to read car bytes for cid ${cid.toString()}`)
+    }
+    // @todo stream car reading
+    const carBytes = new Uint8Array(await carBytesR2.arrayBuffer())
+    const delegations = bytesToDelegations(carBytes)
+    if (delegations.length !== 1) {
+      throw new Error(
+        `expected 1 delegation in CAR, but got ${delegations.length}`
+      )
+    }
+    const [delegation] = delegations
+    return delegation
+  }
+}
+
+/**
+ * @typedef {import('../types/access-api-cf-db').DelegationsV3Table} DelegationsV3Table
+ * @typedef {import('../types/access-api-cf-db').DelegationsV2Table} DelegationsV2Table
+ */
+
+/**
+ * @template {string} TableName
+ * @template {Record<TableName, DelegationsV3Table|DelegationsV2Table>} Tables
+ * @param {import('../types/database').Database<Tables>} db
+ * @param {TableName} delegationsTable
+ * @returns {Promise<bigint>} - count of table
+ */
+async function count(db, delegationsTable) {
+  const { size } = await db
+    .selectFrom(delegationsTable)
+    .select((e) => e.fn.count('cid').as('size'))
+    .executeTakeFirstOrThrow()
+  return BigInt(size)
+}
+
+/**
+ * @param {Ucanto.Delegation} ucan
+ */
+function delegationCarFileKeyer(ucan) {
+  return `${ucan.cid.toString()}.car`
+}
+
+/**
+ * @param {R2Bucket} bucket
+ * @param {Iterable<Ucanto.Delegation>} delegations
+ * @param {(d: Ucanto.Delegation) => string} keyer - builds k/v key strings for each delegation
+ */
+async function writeDelegations(
+  bucket,
+  delegations,
+  keyer = delegationCarFileKeyer
+) {
+  return writeEntries(
+    bucket,
+    [...delegations].map((delegation) => {
+      const key = delegation.cid.toString()
+      const carBytes = delegationsToBytes([delegation])
+      const value = carBytes
+      return /** @type {[key: string, value: Uint8Array]} */ ([key, value])
+    })
+  )
+}
+
+/**
+ * @param {R2Bucket} bucket
+ * @param {Iterable<readonly [key: string, value: Uint8Array ]>} entries
+ */
+async function writeEntries(bucket, entries) {
+  await Promise.all(
+    [...entries].map(async ([key, value]) => {
+      return bucket.put(key, value)
+    })
+  )
+}