feat: write DUDEWHERE link and SATNAV car indexes. (#2238)

- write a CARv2 index so we can use the `SATNAV` bucket to find where
Blocks are.
- write a key to map root CID to car CID so we can use the `DUDEWHERE`
bucket to find where our CAR is in `CARPARK` (for a given root CID).

This enables [freeway](https://github.com/web3-storage/freeway) to
lookup CAR CID(s) for a given root data CID and UnixFS export data
directly from R2.

port of web3-storage/web3.storage#2035

License: MIT
Signed-off-by: Oli Evans <oli@protocol.ai>

Author: olizilla

packages/api/package.json

@@ -27,6 +27,7 @@
     "@nftstorage/ipfs-cluster": "^5.0.1",
     "@noble/ed25519": "^1.6.1",
     "@supabase/postgrest-js": "^0.34.1",
+    "cardex": "^1.0.0",
     "ipfs-car": "^0.6.1",
     "it-last": "^2.0.0",
     "merge-options": "^3.0.4",

packages/api/src/bindings.d.ts

@@ -34,7 +34,13 @@ export interface ServiceConfiguration {
   /** Salt for API key generation */
   SALT: string
 
-  /** R2Bucket binding */
+  /** R2Bucket for CARv2 indexes mapping block offsets within CAR files. */
+  SATNAV: R2Bucket
+
+  /** R2Bucket mapping root data CIDs to CAR CID(s). */
+  DUDEWHERE: R2Bucket
+
+  /** R2Bucket for CAR files */
   CARPARK: R2Bucket
 
   /** Public URL prefix for CARPARK R2 Bucket */

packages/api/src/config.js

@@ -59,6 +59,8 @@ export function serviceConfigFromVariables(vars) {
     MAINTENANCE_MODE: maintenanceModeFromString(vars.MAINTENANCE_MODE),
 
     SALT: vars.SALT,
+    SATNAV: vars.SATNAV,
+    DUDEWHERE: vars.DUDEWHERE,
     CARPARK: vars.CARPARK,
     CARPARK_URL: vars.CARPARK_URL,
     DATABASE_URL: vars.DATABASE_URL,
@@ -108,6 +110,8 @@ export function loadConfigVariables() {
     'ENV',
     'DEBUG',
     'SALT',
+    'SATNAV',
+    'DUDEWHERE',
     'CARPARK',
     'CARPARK_URL',
     'DATABASE_URL',

packages/api/src/utils/car.js

@@ -1,6 +1,8 @@
 import { CID } from 'multiformats/cid'
 import { CarWriter } from '@ipld/car'
 import { sha256 } from 'multiformats/hashes/sha2'
+import { CarIndexer } from '@ipld/car/indexer'
+import { MultihashIndexSortedWriter } from 'cardex'
 
 /**
  * @typedef {import('multiformats/block').Block<unknown>} Block
@@ -39,3 +41,23 @@ export const CAR_CODE = 0x202
 export async function createCarCid(carBytes) {
   return CID.createV1(CAR_CODE, await sha256.digest(carBytes))
 }
+
+/**
+ * @param {Uint8Array} carBytes
+ */
+export async function createCarIndex(carBytes) {
+  const indexer = await CarIndexer.fromBytes(carBytes)
+  const { writer, out } = MultihashIndexSortedWriter.create()
+
+  for await (const blockIndexData of indexer) {
+    // @ts-expect-error CID impl conflict
+    writer.put(blockIndexData)
+  }
+  writer.close()
+
+  const chunks = []
+  for await (const chunk of out) {
+    chunks.push(chunk)
+  }
+  return new Uint8Array(await new Blob(chunks).arrayBuffer())
+}

packages/api/src/utils/context.js

@@ -27,7 +27,12 @@ export async function getContext(event, params) {
     { endpoint: config.S3_ENDPOINT, appName: 'nft' }
   )
 
-  const r2Uploader = new R2Uploader(config.CARPARK, config.CARPARK_URL)
+  const r2Uploader = new R2Uploader({
+    carpark: config.CARPARK,
+    publicUrl: config.CARPARK_URL,
+    dudewhere: config.DUDEWHERE,
+    satnav: config.SATNAV,
+  })
 
   const linkdexApi = config.LINKDEX_URL
     ? new LinkdexApi(config.LINKDEX_URL)

packages/api/src/utils/uploader/r2-uploader.js

@@ -1,27 +1,28 @@
 import pRetry from 'p-retry'
+import { sha256 } from 'multiformats/hashes/sha2'
 import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
+import { createCarIndex } from '../car.js'
 
 /**
  * @typedef {import('../../bindings').Uploader} Uploader
  * @implements {Uploader}
  */
 export class R2Uploader {
   /**
-   * @param {R2Bucket} bucket
-   * @param {string} publicUrl
+   * @param {object} config
+   * @param {R2Bucket} config.carpark
+   * @param {R2Bucket} config.dudewhere
+   * @param {R2Bucket} config.satnav
+   * @param {string} config.publicUrl
    */
-  constructor(bucket, publicUrl) {
-    if (!bucket) throw new Error('missing R2 bucket')
-    if (!publicUrl) throw new Error('missing public url for R2 bucket')
-
-    /**
-     * @private
-     */
-    this._bucket = bucket
-
-    /**
-     * @private
-     */
+  constructor({ carpark, dudewhere, satnav, publicUrl }) {
+    if (!carpark) throw new Error('missing carpark R2 bucket')
+    if (!dudewhere) throw new Error('missing dudewhere R2 bucket')
+    if (!satnav) throw new Error('missing satnav R2 bucket')
+    if (!publicUrl) throw new Error('missing public url for carpark R2 bucket')
+    this._carpark = carpark
+    this._dudewhere = dudewhere
+    this._satnav = satnav
     this._publicUrl = new URL(publicUrl)
   }
 
@@ -42,14 +43,55 @@ export class R2Uploader {
       customMetadata: metadata,
     }
 
-    const put = () => this._bucket.put(key, carBytes, opts)
+    const put = () => this._carpark.put(key, carBytes, opts)
 
     try {
       await pRetry(put, { retries: 3, onFailedAttempt: console.log })
+      await Promise.all([
+        this.uploadDudeWhereLink(metadata.rootCid, carCid),
+        this.uploadSatNavIndex(carBytes, carCid),
+      ])
       return { key, url }
     } catch (cause) {
       // @ts-expect-error wen ts understand Error object?
       throw new Error('Failed to upload CAR to R2', { cause })
     }
   }
+
+  /**
+   * Write a CARv2 index for the passed CAR file to R2.
+   * @param {import('multiformats').CID} carCid
+   * @param {Uint8Array} carBytes
+   */
+  async uploadSatNavIndex(carBytes, carCid) {
+    const indexBytes = await createCarIndex(carBytes)
+    const digest = await sha256.encode(indexBytes)
+    const key = `${carCid}/${carCid}.car.idx`
+    const opts = { sha256: uint8ArrayToString(digest, 'base16') }
+    try {
+      return await pRetry(async () => this._satnav.put(key, indexBytes, opts), {
+        retries: 3,
+      })
+    } catch (cause) {
+      // @ts-expect-error error.cause is legit.
+      throw new Error('Failed to write satnav index to R2', { cause })
+    }
+  }
+
+  /**
+   * @param {string} rootCid
+   * @param {import('multiformats').CID} carCid
+   */
+  async uploadDudeWhereLink(rootCid, carCid) {
+    const key = `${rootCid}/${carCid}`
+    const data = new Uint8Array()
+    try {
+      return await pRetry(async () => this._dudewhere.put(key, data), {
+        retries: 3,
+      })
+    } catch (cause) {
+      // @ts-expect-error error.cause is legit.
+      throw new Error('Failed to write dudewhere index to R2', { cause })
+    }
+  }
 }

packages/api/test/config.spec.js

@@ -36,6 +36,8 @@ const BASE_CONFIG = {
   CLUSTER_API_URL: 'http://127.0.0.1:9094',
   S3_ENDPOINT: 'http://127.0.0.1:9000',
   SLACK_USER_REQUEST_WEBHOOK_URL: '',
+  SATNAV: '?',
+  DUDEWHERE: '?',
   CARPARK: '?',
   CARPARK_URL: 'http://example.org',
   LINKDEX_URL: 'http://example.org',

packages/api/test/nfts-upload.spec.js

@@ -1,12 +1,13 @@
 import test from 'ava'
-import pRetry from 'p-retry'
 import { CID } from 'multiformats/cid'
 import * as Block from 'multiformats/block'
 import { sha256, sha512 } from 'multiformats/hashes/sha2'
 import * as pb from '@ipld/dag-pb'
 import { CarWriter } from '@ipld/car'
 import { packToBlob } from 'ipfs-car/pack/blob'
+import { MultihashIndexSortedReader } from 'cardex'
 import { TreewalkCarSplitter } from 'carbites/treewalk'
+import { equals as uint8ArrayEquals } from 'uint8arrays/equals'
 import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
 import { createClientWithUser, getRawClient } from './scripts/helpers.js'
 import { createCar } from './scripts/car.js'
@@ -21,6 +22,7 @@ import {
 import { File } from 'nft.storage/src/platform.js'
 import crypto from 'node:crypto'
 import { FormData } from 'undici'
+import { createCarCid } from '../src/utils/car.js'
 
 test.before(async (t) => {
   const linkdexUrl = 'http://fake.api.net'
@@ -735,6 +737,75 @@ test.serial('should update a single file', async (t) => {
   t.is(uploadData.name, name)
 })
 
+test.serial('should write satnav index', async (t) => {
+  const client = await createClientWithUser(t)
+  const config = getTestServiceConfig(t)
+  const mf = getMiniflareContext(t)
+  const { root, car: carBody } = await createCar('satnav')
+  const carBytes = new Uint8Array(await carBody.arrayBuffer())
+  const carCid = await createCarCid(carBytes)
+
+  const res = await mf.dispatchFetch('http://miniflare.test/upload', {
+    method: 'POST',
+    headers: {
+      Authorization: `Bearer ${client.token}`,
+      'Content-Type': 'application/car',
+    },
+    body: carBody,
+  })
+
+  const { ok, value } = await res.json()
+  t.truthy(ok, 'Server response payload has `ok` property')
+  t.is(value.cid, root.toString(), 'Server responded with expected CID')
+  t.is(value.type, 'application/car', 'type should match car mime-type')
+
+  const r2Bucket = await mf.getR2Bucket('SATNAV')
+  const r2Object = await r2Bucket.get(`${carCid}/${carCid}.car.idx`)
+  if (!r2Object?.body) {
+    t.fail('repsonse stream must exist')
+  }
+  // @ts-expect-error
+  const reader = MultihashIndexSortedReader.fromIterable(r2Object?.body)
+  const entries = []
+  for await (const entry of reader.entries()) {
+    entries.push(entry)
+  }
+
+  t.is(entries.length, 1, 'Index contains a single entry')
+  t.true(
+    uint8ArrayEquals(entries[0].digest, root.multihash.digest),
+    'Index entry is for root data CID'
+  )
+})
+
+test.serial('should write dudewhere index', async (t) => {
+  const client = await createClientWithUser(t)
+  const config = getTestServiceConfig(t)
+  const mf = getMiniflareContext(t)
+  const { root, car: carBody } = await createCar('dude')
+  const carBytes = new Uint8Array(await carBody.arrayBuffer())
+  const carCid = await createCarCid(carBytes)
+
+  const res = await mf.dispatchFetch('http://miniflare.test/upload', {
+    method: 'POST',
+    headers: {
+      Authorization: `Bearer ${client.token}`,
+      'Content-Type': 'application/car',
+    },
+    body: carBody,
+  })
+
+  const { ok, value } = await res.json()
+  t.truthy(ok, 'Server response payload has `ok` property')
+  t.is(value.cid, root.toString(), 'Server responded with expected CID')
+  t.is(value.type, 'application/car', 'type should match car mime-type')
+
+  const r2Bucket = await mf.getR2Bucket('DUDEWHERE')
+  const r2Objects = await r2Bucket.list({ prefix: `${root}/` })
+  t.is(r2Objects.objects.length, 1)
+  t.is(r2Objects.objects[0].key, `${root}/${carCid}`)
+})
+
 /**
  * @param {Uint8Array} data
  */

packages/api/wrangler.toml

@@ -9,17 +9,21 @@ workers_dev = true
 compatibility_date = "2021-08-23"
 compatibility_flags = ["formdata_parser_supports_files"]
 no_bundle = true
+r2_buckets = [
+   # CAR files live here.
+   { binding = "CARPARK", bucket_name = "carpark-staging-0" },
+   # Mapping root data CIDs to CAR CID(s).
+   { binding = "DUDEWHERE", bucket_name = "dudewhere-staging-0" },
+   # CARv2 indexes - a mapping of block offsets within CAR files.
+   { binding = "SATNAV", bucket_name = "satnav-staging-0" }
+]
 
 [vars]
 ENV = "dev"
 DEBUG = "true"
 DATABASE_URL = "http://localhost:3000"
 CARPARK_URL = "https://carpark-dev.web3.storage"
 
-[[r2_buckets]]
-binding = 'CARPARK'
-bucket_name = 'carpark-dev-0'
-
 [build]
 command = "scripts/cli.js build"
 
@@ -29,17 +33,18 @@ command = "scripts/cli.js build"
 name = "nft-storage-staging"
 route = "api-staging.nft.storage/*"
 usage_model = "unbound"
+r2_buckets = [
+   { binding = "CARPARK", bucket_name = "carpark-staging-0" },
+   { binding = "DUDEWHERE", bucket_name = "dudewhere-staging-0" },
+   { binding = "SATNAV", bucket_name = "satnav-staging-0" }
+ ]
 
 [env.staging.vars]
 ENV = "staging"
 DEBUG = "true"
 DATABASE_URL = "https://nft-storage-pgrest-staging.herokuapp.com"
 CARPARK_URL = "https://carpark-staging.web3.storage"
 
-[[env.staging.r2_buckets]]
-binding = 'CARPARK'
-bucket_name = 'carpark-staging-0'
-
 [env.staging.build]
 command = "scripts/cli.js build --env staging"
 watch_dir = "src"
@@ -53,17 +58,18 @@ format = "service-worker"
 name = "nft-storage"
 route = "api.nft.storage/*"
 usage_model = "unbound"
+r2_buckets = [
+   { binding = "CARPARK", bucket_name = "carpark-prod-0" },
+   { binding = "DUDEWHERE", bucket_name = "dudewhere-prod-0" },
+   { binding = "SATNAV", bucket_name = "satnav-prod-0" }
+ ]
 
 [env.production.vars]
 ENV = "production"
 DEBUG = "false"
 DATABASE_URL = "https://nft-storage-pgrest-prod.herokuapp.com"
 CARPARK_URL = "https://carpark.web3.storage"
 
-[[env.production.r2_buckets]]
-binding = 'CARPARK'
-bucket_name = 'carpark-prod-0'
-
 [env.production.build]
 command = "scripts/cli.js build --env production"
 watch_dir = "src"