feat: store pins in datastore instead of a DAG (#2771)

Adds a `.pins` datastore to `ipfs-repo` and uses that to store pins as cbor binary keyed by multihash.

### Format

As stored in the datastore, each pin has several fields:

```javascript
{
  codec: // optional Number, the codec from the CID that this multihash was pinned with, if omitted, treated as 'dag-pb'
  version: // optional Number, the version number from the CID that this multihash was pinned with, if omitted, treated as v0
  depth: // Number Infinity = recursive pin, 0 = direct, 1+ = pinned to a depth
  comments: // optional String user-friendly description of the pin
  metadata: // optional Object, user-defined data for the pin
}
```

Notes:

`.codec` and `.version` are stored so we can recreate the original CID when listing pins.

### Metadata

The intention is for us to be able to add extra fields that have technical meaning to the root of the object, and the user can store application-specific data in the `metadata` field.

### CLI

```console
$ ipfs pin add bafyfoo --metadata key1=value1,key2=value2
$ ipfs pin add bafyfoo --metadata-format=json --metadata '{"key1":"value1","key2":"value2"}'

$ ipfs pin list
bafyfoo

$ ipfs pin list -l
CID      Name    Type       Metadata
bafyfoo  My pin  Recursive  {"key1":"value1","key2":"value2"}

$ ipfs pin metadata Qmfoo --format=json
{"key1":"value1","key2":"value2"}
```

### HTTP API

* '/api/v0/pin/add' route adds new `metadata` argument, accepts a json string
* '/api/v0/pin/metadata' returns metadata as json

### Core API

* `ipfs.pin.addAll` accepts and returns an async iterator
* `ipfs.pin.rmAll` accepts and returns an async iterator

```javascript
// pass a cid or IPFS Path with options
const { cid } = await ipfs.pin.add(new CID('/ipfs/Qmfoo'), {
  recursive: false,
  metadata: {
    key: 'value
  },
  timeout: 2000
}))

// pass an iterable of CIDs
const [{ cid: cid1 }, { cid: cid2 }] = await all(ipfs.pin.addAll([
  new CID('/ipfs/Qmfoo'),
  new CID('/ipfs/Qmbar')
], { timeout: '2s' }))

// pass an iterable of objects with options
const [{ cid: cid1 }, { cid: cid2 }] = await all(ipfs.pin.addAll([
  { cid: new CID('/ipfs/Qmfoo'), recursive: true, comments: 'A recursive pin' },
  { cid: new CID('/ipfs/Qmbar'), recursive: false, comments: 'A direct pin' }
], { timeout: '2s' }))
```

* ipfs.pin.rmAll accepts and returns an async generator (other input types are available)

```javascript
// pass an IPFS Path or CID
const { cid } = await ipfs.rm(new CID('/ipfs/Qmfoo/file.txt'))

// pass options
const { cid } = await all(ipfs.rm(new CID('/ipfs/Qmfoo'), { recursive: true }))

// pass an iterable of CIDs or objects with options
const [{ cid }] = await all(ipfs.rmAll([{ cid: new CID('/ipfs/Qmfoo'), recursive: true }]))
```

Bonus: Lets us pipe the output of one command into another:

```javascript
await pipe(
	ipfs.pin.ls({ type: 'recursive' }),
    (source) => ipfs.pin.rmAll(source)
)

// or
await all(ipfs.pin.rmAll(ipfs.pin.ls({ type: 'recursive'})))
```

BREAKING CHANGES:

* pins are now stored in a datastore, a repo migration will occur on startup
* All deps of this module now use Uint8Arrays in place of node Buffers

Author: achingbrain

package.json

@@ -43,38 +43,39 @@
     "abort-controller": "^3.0.0",
     "any-signal": "^1.1.0",
     "bignumber.js": "^9.0.0",
-    "buffer": "^5.6.0",
-    "cids": "^0.8.3",
+    "cids": "^1.0.0",
     "debug": "^4.1.0",
     "form-data": "^3.0.0",
     "ipfs-core-utils": "^0.3.2",
     "ipfs-utils": "^3.0.0",
-    "ipld-block": "^0.9.2",
-    "ipld-dag-cbor": "^0.16.0",
-    "ipld-dag-pb": "^0.19.0",
-    "ipld-raw": "^5.0.0",
+    "ipld-block": "^0.10.0",
+    "ipld-dag-cbor": "^0.17.0",
+    "ipld-dag-pb": "^0.20.0",
+    "ipld-raw": "^6.0.0",
     "iso-url": "^0.4.7",
     "it-last": "^1.0.1",
+    "it-map": "^1.0.2",
     "it-tar": "^1.2.2",
     "it-to-buffer": "^1.0.0",
     "it-to-stream": "^0.1.1",
     "merge-options": "^2.0.0",
-    "multiaddr": "^7.4.3",
-    "multiaddr-to-uri": "^5.1.0",
-    "multibase": "^1.0.1",
-    "multicodec": "^1.0.0",
-    "multihashes": "^1.0.1",
+    "multiaddr": "^8.0.0",
+    "multiaddr-to-uri": "^6.0.0",
+    "multibase": "^3.0.0",
+    "multicodec": "^2.0.0",
+    "multihashes": "^3.0.1",
     "nanoid": "^3.0.2",
     "node-fetch": "^2.6.0",
     "parse-duration": "^0.4.4",
-    "stream-to-it": "^0.2.1"
+    "stream-to-it": "^0.2.1",
+    "uint8arrays": "^1.1.0"
   },
   "devDependencies": {
-    "aegir": "^23.0.0",
+    "aegir": "^26.0.0",
     "cross-env": "^7.0.0",
     "go-ipfs": "^0.6.0",
     "interface-ipfs-core": "^0.139.1",
-    "ipfsd-ctl": "^5.0.0",
+    "ipfsd-ctl": "^6.0.0",
     "it-all": "^1.0.1",
     "it-concat": "^1.0.0",
     "it-pipe": "^1.1.0",

src/block/get.js

@@ -2,7 +2,6 @@
 
 const Block = require('ipld-block')
 const CID = require('cids')
-const { Buffer } = require('buffer')
 const configure = require('../lib/configure')
 const toUrlSearchParams = require('../lib/to-url-search-params')
 
@@ -20,6 +19,6 @@ module.exports = configure(api => {
       headers: options.headers
     })
 
-    return new Block(Buffer.from(await res.arrayBuffer()), cid)
+    return new Block(new Uint8Array(await res.arrayBuffer()), cid)
   }
 })

src/config/replace.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const { Buffer } = require('buffer')
+const uint8ArrayFromString = require('uint8arrays/from-string')
 const multipartRequest = require('../lib/multipart-request')
 const configure = require('../lib/configure')
 const toUrlSearchParams = require('../lib/to-url-search-params')
@@ -18,7 +18,7 @@ module.exports = configure(api => {
       signal,
       searchParams: toUrlSearchParams(options),
       ...(
-        await multipartRequest(Buffer.from(JSON.stringify(config)), controller, options.headers)
+        await multipartRequest(uint8ArrayFromString(JSON.stringify(config)), controller, options.headers)
       )
     })
 

src/dag/get.js

@@ -18,16 +18,16 @@ module.exports = configure((api, options) => {
   return async (cid, options = {}) => {
     const resolved = await dagResolve(cid, options)
     const block = await getBlock(resolved.cid, options)
-    const dagResolver = resolvers[block.cid.codec]
+    const dagResolver = resolvers[resolved.cid.codec]
 
     if (!dagResolver) {
       throw Object.assign(
-        new Error(`Missing IPLD format "${block.cid.codec}"`),
-        { missingMulticodec: cid.codec }
+        new Error(`Missing IPLD format "${resolved.cid.codec}"`),
+        { missingMulticodec: resolved.cid.codec }
       )
     }
 
-    if (block.cid.codec === 'raw' && !resolved.remPath) {
+    if (resolved.cid.codec === 'raw' && !resolved.remPath) {
       resolved.remainderPath = '/'
     }
 

src/dht/find-peer.js

@@ -1,6 +1,5 @@
 'use strict'
 
-const { Buffer } = require('buffer')
 const CID = require('cids')
 const multiaddr = require('multiaddr')
 const configure = require('../lib/configure')
@@ -13,7 +12,7 @@ module.exports = configure(api => {
       timeout: options.timeout,
       signal: options.signal,
       searchParams: toUrlSearchParams({
-        arg: `${Buffer.isBuffer(peerId) ? new CID(peerId) : peerId}`,
+        arg: `${peerId instanceof Uint8Array ? new CID(peerId) : peerId}`,
         ...options
       }),
       headers: options.headers

src/dht/get.js

@@ -1,25 +1,26 @@
 'use strict'
 
-const { Buffer } = require('buffer')
 const configure = require('../lib/configure')
 const toUrlSearchParams = require('../lib/to-url-search-params')
 const { Value } = require('./response-types')
+const uint8ArrayToString = require('uint8arrays/to-string')
+const uint8ArrayFromString = require('uint8arrays/from-string')
 
 module.exports = configure(api => {
   return async function get (key, options = {}) {
     const res = await api.post('dht/get', {
       timeout: options.timeout,
       signal: options.signal,
       searchParams: toUrlSearchParams({
-        arg: Buffer.isBuffer(key) ? key.toString() : key,
+        arg: key instanceof Uint8Array ? uint8ArrayToString(key) : key,
         ...options
       }),
       headers: options.headers
     })
 
     for await (const message of res.ndjson()) {
       if (message.Type === Value) {
-        return Buffer.from(message.Extra, 'base64')
+        return uint8ArrayFromString(message.Extra, 'base64pad')
       }
     }
 

src/files/read.js

@@ -1,6 +1,5 @@
 'use strict'
 
-const { Buffer } = require('buffer')
 const toIterable = require('stream-to-it/source')
 const configure = require('../lib/configure')
 const toUrlSearchParams = require('../lib/to-url-search-params')
@@ -18,8 +17,6 @@ module.exports = configure(api => {
       headers: options.headers
     })
 
-    for await (const chunk of toIterable(res.body)) {
-      yield Buffer.from(chunk)
-    }
+    yield * toIterable(res.body)
   }
 })

src/get.js

@@ -1,18 +1,18 @@
 'use strict'
 
 const Tar = require('it-tar')
-const { Buffer } = require('buffer')
 const CID = require('cids')
 const configure = require('./lib/configure')
 const toUrlSearchParams = require('./lib/to-url-search-params')
+const map = require('it-map')
 
 module.exports = configure(api => {
   return async function * get (path, options = {}) {
     const res = await api.post('get', {
       timeout: options.timeout,
       signal: options.signal,
       searchParams: toUrlSearchParams({
-        arg: `${Buffer.isBuffer(path) ? new CID(path) : path}`,
+        arg: `${path instanceof Uint8Array ? new CID(path) : path}`,
         ...options
       }),
       headers: options.headers
@@ -28,7 +28,7 @@ module.exports = configure(api => {
       } else {
         yield {
           path: header.name,
-          content: body
+          content: map(body, (chunk) => chunk.slice()) // convert bl to Buffer/Uint8Array
         }
       }
     }

src/index.js

@@ -1,6 +1,6 @@
 'use strict'
 /* eslint-env browser */
-const { Buffer } = require('buffer')
+
 const CID = require('cids')
 const multiaddr = require('multiaddr')
 const multibase = require('multibase')
@@ -56,6 +56,6 @@ function ipfsClient (options = {}) {
   }
 }
 
-Object.assign(ipfsClient, { Buffer, CID, multiaddr, multibase, multicodec, multihash, globSource, urlSource })
+Object.assign(ipfsClient, { CID, multiaddr, multibase, multicodec, multihash, globSource, urlSource })
 
 module.exports = ipfsClient

src/ls.js

@@ -1,6 +1,5 @@
 'use strict'
 
-const { Buffer } = require('buffer')
 const CID = require('cids')
 const configure = require('./lib/configure')
 const toUrlSearchParams = require('./lib/to-url-search-params')
@@ -11,7 +10,7 @@ module.exports = configure(api => {
       timeout: options.timeout,
       signal: options.signal,
       searchParams: toUrlSearchParams({
-        arg: `${Buffer.isBuffer(path) ? new CID(path) : path}`,
+        arg: `${path instanceof Uint8Array ? new CID(path) : path}`,
         ...options
       }),
       headers: options.headers