feat: post car api (#25)

This PR adds `POST /car` to the API.

packages/api/README.md

@@ -10,6 +10,21 @@ One time set up of your cloudflare worker subdomain for dev:
 - Sign up to Cloudflare and log in with your default browser.
 - `npm i @cloudflare/wrangler -g` - Install the Cloudflare wrangler CLI
 - `wrangler login` - Authenticate your wrangler cli; it'll open your browser.
+- Setup Cluster
+    - You need to run a cluster locally and make it accessible from the internet for development.
+    - Follow the quickstart guide to get an IPFS Cluster up and running: https://cluster.ipfs.io/documentation/quickstart/
+    - Install [localtunnel](https://theboroer.github.io/localtunnel-www/) and expose the IPFS Cluster HTTP API and IPFS Proxy API (replacing "USER" with your name):
+
+    ```sh
+    npm install -g localtunnel
+    lt --port 9094 --subdomain USER-cluster-api-web3-storage
+    ```
+
+    - There is an npm script you can use to quickly establish these tunnels during development:
+
+    ```sh
+    npm run lt
+    ```
 - Copy your cloudflare account id from `wrangler whoami`
 - Update `wrangler.toml` with a new `env`. Set your env name to be the value of `whoami` on your system you can use `npm start` to run the worker in dev mode for you.
 
@@ -19,6 +34,7 @@ One time set up of your cloudflare worker subdomain for dev:
     [env.bobbytables]
     workers_dev = true
     account_id = "<what does the `wrangler whoami` say>"
+    vars = { CLUSTER_API_URL = "https://USER-cluster-api-web3-storage.loca.lt" }
     ```
 
 - `npm run build` - Build the bundle
@@ -27,7 +43,8 @@ One time set up of your cloudflare worker subdomain for dev:
     ```sh
     wrangler secret put MAGIC_SECRET_KEY --env $(whoami) # Get from magic.link account
     wrangler secret put SALT --env $(whoami) # open `https://csprng.xyz/v1/api` in the browser and use the value of `Data`
-    wrangler secret put FAUNA_KEY --env $(whoami) # Get from fauna.com after creating a dev DB
+    wrangler secret put FAUNA_KEY --env $(whoami) # Get from fauna.com after creating a dev Classic DB
+    wrangler secret put CLUSTER_BASIC_AUTH_TOKEN --env $(whoami) # Get from web3.storage vault in 1password (not required for dev)
     ```
 
 - `npm run publish` - Publish the worker under your env. An alias for `wrangler publish --env $(whoami)`

packages/api/package.json

@@ -6,24 +6,30 @@
   "license": "(Apache-2.0 AND MIT)",
   "main": "dist/main.js",
   "scripts": {
+    "lt": "npm run lt:cluster",
+    "lt:cluster": "npx localtunnel --port 9094 --subdomain \"$(whoami)-cluster-api-web3-storage\"",
     "start": "wrangler dev --env $(whoami)",
     "dev": "wrangler dev --env $(whoami)",
     "publish": "wrangler publish --env $(whoami)",
     "build": "webpack",
     "format": "prettier --write '**/*.{js,css,json,md}'",
-    "test": "npm-run-all -p -r mock:db test:e2e",
+    "test": "npm-run-all -p -r mock:cluster mock:db test:e2e",
     "test:e2e": "playwright-test \"test/**/*.spec.js\" --sw src/index.js",
+    "mock:cluster": "smoke -p 9094 test/mocks/cluster",
     "mock:db": "smoke -p 9086 test/mocks/db"
   },
   "devDependencies": {
+    "@ipld/car": "^3.1.4",
+    "@nftstorage/ipfs-cluster": "^3.1.2",
     "@types/mocha": "^8.2.2",
     "assert": "^2.0.0",
     "buffer": "^6.0.3",
+    "ipfs-car": "^0.4.3",
     "npm-run-all": "^4.1.5",
     "playwright-test": "^5.0.0",
     "process": "^0.11.10",
-    "stream-browserify": "^3.0.0",
     "smoke": "^3.1.1",
+    "stream-browserify": "^3.0.0",
     "webpack": "^5.42.0",
     "webpack-cli": "^4.7.2"
   },

packages/api/src/car.js

@@ -1,6 +1,9 @@
+import { gql } from '@web3-storage/db'
 import { GATEWAY } from './constants.js'
 import { JSONResponse } from './utils/json-response.js'
 
+const LOCAL_ADD_THRESHOLD = 1024 * 1024 * 2.5
+
 // TODO: ipfs should let us ask the size of a CAR file.
 // This consumes the CAR response from ipfs to find the content-length.
 export async function carHead(request, env, ctx) {
@@ -19,44 +22,86 @@ export async function carHead(request, env, ctx) {
 export async function carGet(request, env, ctx) {
   const cache = caches.default
   let res = await cache.match(request)
-  if (!res) {
-    const {
-      params: { cid },
-    } = request
-    // gateway does not support `carversion` yet.
-    // using it now means we can skip the cache if it is supported in the future
-    const url = new URL(`/api/v0/dag/export?arg=${cid}&carversion=1`, GATEWAY)
-    console.log(url.toString())
-    res = await fetch(url, { method: 'POST' })
-    if (!res.ok) {
-      // bail early. dont cache errors.
-      return res
-    }
-    // TODO: these headers should be upstreamed to ipfs impls
-    // without the content-disposition, firefox describes them as DMS files.
-    const headers = {
-      'Cache-Control': 'public, max-age=31536000',  // max max-age is 1 year
-      'Content-Type': 'application/car',
-      'Content-Disposition': `attachment; filename="${cid}.car"`,
-    }
-    // // compress if asked for? is it worth it?
-    // if (request.headers.get('Accept-Encoding').match('gzip')) {
-    //   headers['Content-Encoding'] = 'gzip'
-    // }
-    res = new Response(res.body, { ...res, headers })
-    ctx.waitUntil(cache.put(request, res.clone()))
+
+  if (res) {
+    return res
   }
+
+  const {
+    params: { cid },
+  } = request
+  // gateway does not support `carversion` yet.
+  // using it now means we can skip the cache if it is supported in the future
+  const url = new URL(`/api/v0/dag/export?arg=${cid}&carversion=1`, GATEWAY)
+  console.log(url.toString())
+  res = await fetch(url, { method: 'POST' })
+  if (!res.ok) {
+    // bail early. dont cache errors.
+    return res
+  }
+  // TODO: these headers should be upstreamed to ipfs impls
+  // without the content-disposition, firefox describes them as DMS files.
+  const headers = {
+    'Cache-Control': 'public, max-age=31536000',  // max max-age is 1 year
+    'Content-Type': 'application/car',
+    'Content-Disposition': `attachment; filename="${cid}.car"`,
+  }
+  // // compress if asked for? is it worth it?
+  // if (request.headers.get('Accept-Encoding').match('gzip')) {
+  //   headers['Content-Encoding'] = 'gzip'
+  // }
+  res = new Response(res.body, { ...res, headers })
+  ctx.waitUntil(cache.put(request, res.clone()))
+
   return res
 }
 
 /**
+ * Post a CAR file.
+ *
  * @param {import('./user').AuthenticatedRequest} request
+ * @param {import('./env').Env} env
  */
-export async function carPost(request, env, ctx) {
-  const { user, authToken } = request.auth
-  console.log('User ID', user._id)
-  console.log('Auth Token ID', authToken && authToken._id)
-  return new JSONResponse({})
+export async function carPost(request, env) {
+  const { _id } = request.auth.authToken
+  const { headers } = request
+
+  let name = headers.get('x-name')
+  if (!name || typeof name !== 'string') {
+    name = `Upload at ${new Date().toISOString()}`
+  }
+
+  const blob = await request.blob()
+
+  // Ensure car blob.type is set; it is used by the cluster client to set the foramt=car flag on the /add call.
+  const content = blob.slice(0, blob.size, 'application/car')
+
+  const { cid } = await env.cluster.add(content, {
+    metadata: { size: content.size.toString() },
+    // When >2.5MB, use local add, because waiting for blocks to be sent to
+    // other cluster nodes can take a long time. Replication to other nodes
+    // will be done async by bitswap instead.
+    local: blob.size > LOCAL_ADD_THRESHOLD,
+  })
+
+  // Store in DB
+  await env.db.query(gql`
+    mutation importCar($data: ImportCarInput!) {
+      importCar(data: $data) {
+        name
+      }
+    }
+  `, {
+    data: { 
+      authToken: _id,
+      cid,
+      name
+      // dagSize: undefined // TODO: should we default to chunk car behavior?
+    }
+  })
+
+  // TODO: Improve response type with pin information
+  return new JSONResponse({ cid })
 }
 
 export async function carPut(request, env, ctx) {

packages/api/src/env.js

@@ -1,5 +1,6 @@
 import { Magic } from '@magic-sdk/admin'
 import { DBClient } from '@web3-storage/db'
+import { Cluster } from '@nftstorage/ipfs-cluster'
 
 /** @typedef {{ magic: Magic, db: DBClient, SALT: string }} Env */
 
@@ -11,9 +12,13 @@ export function envAll (_, env) {
   env.magic = new Magic(env.MAGIC_SECRET_KEY || MAGIC_SECRET_KEY)
 
   env.db = new DBClient({
-    endpoint: env.FAUNA_ENDPOINT || FAUNA_ENDPOINT,
+    endpoint: env.FAUNA_ENDPOINT || (typeof FAUNA_ENDPOINT === 'undefined' ? undefined : FAUNA_ENDPOINT),
     token: env.FAUNA_KEY || FAUNA_KEY
   })
 
   env.SALT = env.SALT || SALT
+
+  const clusterAuthToken = env.CLUSTER_BASIC_AUTH_TOKEN || (typeof CLUSTER_BASIC_AUTH_TOKEN === 'undefined' ? undefined : CLUSTER_BASIC_AUTH_TOKEN)
+  const headers = clusterAuthToken ? { Authorization: `Basic ${clusterAuthToken}` } : null
+  env.cluster = new Cluster(env.CLUSTER_API_URL || CLUSTER_API_URL, { headers })
 }

packages/api/src/user.js

@@ -100,6 +100,7 @@ export function withAuth (handler) {
    */
   return async (request, env, ctx) => {
     const auth = request.headers.get('Authorization') || ''
+    // TODO: Should this throw if no auth token with meaningful error?
     const token = env.magic.utils.parseAuthorizationHeader(auth)
 
     // validate access tokens

packages/api/test/car.spec.js

@@ -0,0 +1,41 @@
+import assert from 'assert'
+import { endpoint } from './scripts/constants.js'
+import * as JWT from '../src/utils/jwt.js'
+import { SALT } from './scripts/worker-globals.js'
+import { createCar } from './scripts/car.js'
+import { JWT_ISSUER } from '../src/constants.js'
+
+function getTestJWT (sub = 'test', name = 'test') {
+  return JWT.sign({ sub, iss: JWT_ISSUER, iat: Date.now(), name }, SALT)
+}
+
+describe('POST /car', () => {
+  it('should add posted CARs to Cluster', async () => {
+    const name = 'car'
+    // Create token
+    const token = await getTestJWT()
+
+    // Create Car
+    const { root, car: carBody } = await createCar('hello world!')
+
+    // expected CID for the above data
+    const expectedCid = 'bafkreidvbhs33ighmljlvr7zbv2ywwzcmp5adtf4kqvlly67cy56bdtmve'
+    assert.strictEqual(root.toString(), expectedCid, 'car file has correct root')
+
+    const res = await fetch(new URL('car', endpoint).toString(), {
+      method: 'POST',
+      headers: {
+        Authorization: `Bearer ${token}`,
+        'Content-Type': 'application/car',
+        'X-Name': name
+      },
+      body: carBody,
+    })
+
+    assert(res, 'Server responded')
+    assert(res.ok, 'Server response ok')
+    const { cid } = await res.json()
+    assert(cid, 'Server response payload has `cid` property')
+    assert.strictEqual(cid, cid, 'Server responded with expected CID')
+  })
+})

packages/api/test/mocks/cluster/post_add$format=car.js

@@ -0,0 +1,26 @@
+const { CarReader } = require('@ipld/car')
+
+/**
+ * https://github.com/sinedied/smoke#javascript-mocks
+ * @typedef {{ buffer: Buffer, originalname: string }} MultrFile
+ * @param {{ query: Record<string, string>, files: MultrFile[] }} request
+ */
+module.exports = async ({ query, files }) => {
+  const car = await CarReader.fromBytes(files[0].buffer)
+  const roots = await car.getRoots()
+  // @ts-ignore
+  const { cid, bytes } = await car.get(roots[0])
+  const result = {
+    cid: {
+      '/': cid.toString(),
+    },
+    name: files[0].originalname,
+    // car uploads may not be unixfs, so get a bytes property instead of `size` https://github.com/ipfs/ipfs-cluster/issues/1362
+    bytes: bytes.length,
+  }
+  return {
+    statusCode: 200,
+    headers: { 'Content-Type': 'application/json' },
+    body: query['stream-channels'] === 'false' ? [result] : result,
+  }
+}

packages/api/test/mocks/db/post_graphql.js

@@ -8,6 +8,10 @@ const gqlOkResponse = data => gqlResponse(200, { data })
  * @param {{ body: GraphQLRequest }} request
  */
 module.exports = ({ body }) => {
+  if (body.query.includes('importCar')) {
+    return gqlOkResponse({ importCar: { _id: 'test-upload-id' } })
+  }
+
   if (body.query.includes('verifyAuthToken')) {
     return gqlOkResponse({
       verifyAuthToken: {

packages/api/test/scripts/car.js

@@ -0,0 +1,10 @@
+import { packToBlob } from 'ipfs-car/pack/blob'
+
+/**
+ * @param {string} str Data to encode into CAR file.
+ */
+export async function createCar(str) {
+  return await packToBlob({
+    input: new TextEncoder().encode(str)
+  })
+}

packages/api/test/scripts/worker-globals.js

@@ -9,3 +9,5 @@ export const SALT = 'test-salt'
 export const FAUNA_ENDPOINT = 'http://localhost:9086/graphql'
 export const FAUNA_KEY = 'test-fauna-key'
 export const MAGIC_SECRET_KEY = 'test-magic-secret-key'
+export const CLUSTER_API_URL = 'http://localhost:9094'
+export const CLUSTER_BASIC_AUTH_TOKEN = 'test'

packages/api/wrangler.toml

@@ -11,6 +11,8 @@ command = "npm run build"
 watch_dir = "src"
 [build.upload]
 format = "service-worker"
+[vars]
+CLUSTER_API_URL = "https://web3.storage.ipfscluster.io/api/"
 
 # ---- Environment specific overrides below ! ----
 # NOTE: wrangler automatically assigns each env the root `name` with the env name suffixed on the end
@@ -36,18 +38,22 @@ route = "https://api-staging.web3.storage/*"
 [env.alan]
 workers_dev = true
 account_id = "4fe12d085474d33bdcfd8e9bed4d8f95"
+vars = { CLUSTER_API_URL = "https://alan-cluster-api-web3-storage.loca.lt" }
 
 [env.oli]
 workers_dev = true
 account_id = "6e5a2aed80cd37d77e8d0c797a75ebbd"
+vars = { CLUSTER_API_URL = "https://oli-cluster-api-web3-storage.loca.lt" }
 
 [env.yusef]
 workers_dev = true
 account_id = "8c3da25233263bd7a26c0e2e04569ded"
+vars = { CLUSTER_API_URL = "https://yusef-cluster-api-web3-storage.loca.lt" }
 
 [env.vsantos]
 workers_dev = true
 account_id = "7ec0b7cf2ec201b2580374e53ba5f37b"
+vars = { CLUSTER_API_URL = "https://vsantos-cluster-api-web3-storage.loca.lt" }
 
 # Create your env name as the value of `whoami` on your system, so you can run `npm start` to run in dev mode.
 # Copy this template and fill out the values

packages/client/examples/node.js/files.js

@@ -1,4 +1,4 @@
-import { Web3Storage } from '../../src/lib.js'
+import { Web3Storage, File } from '../../src/lib.js'
 
 // TODO
 const endpoint = 'https://api.web3.storage' // the default

packages/client/package.json

@@ -30,13 +30,13 @@
     "clean": "del dist"
   },
   "dependencies": {
-    "@ipld/car": "^3.1.2",
+    "@ipld/car": "^3.1.4",
     "@web-std/blob": "^2.1.0",
     "@web-std/fetch": "^2.0.1",
     "@web-std/file": "^1.1.0",
     "browser-readablestream-to-it": "^1.0.2",
     "carbites": "^1.0.6",
-    "ipfs-car": "^0.4.1",
+    "ipfs-car": "^0.4.3",
     "p-retry": "^4.5.0",
     "streaming-iterables": "^6.0.0"
   },
@@ -48,7 +48,7 @@
     "del-cli": "^4.0.0",
     "hundreds": "0.0.9",
     "mocha": "8.3.2",
-    "multiformats": "^9.1.1",
+    "multiformats": "^9.1.2",
     "npm-run-all": "^4.1.5",
     "nyc": "15.1.0",
     "playwright-test": "^4.1.0",