33const core = require ( 'datastore-core' )
44const ShardingStore = core . ShardingDatastore
55const Block = require ( 'ipld-block' )
6- const { cidToKey, keyToCid } = require ( './blockstore-utils' )
6+ const { cidToKey } = require ( './blockstore-utils' )
77const map = require ( 'it-map' )
8- const pipe = require ( 'it-pipe' )
8+ const drain = require ( 'it-drain' )
9+ const pushable = require ( 'it-pushable' )
910
1011module . exports = async ( filestore , options ) => {
1112 const store = await maybeWithSharding ( filestore , options )
@@ -23,7 +24,7 @@ function maybeWithSharding (filestore, options) {
2324function createBaseStore ( store ) {
2425 return {
2526 /**
26- * Query the store.
27+ * Query the store
2728 *
2829 * @param {Object } query
2930 * @param {Object } options
@@ -32,38 +33,23 @@ function createBaseStore (store) {
3233 async * query ( query , options ) { // eslint-disable-line require-await
3334 yield * store . query ( query , options )
3435 } ,
36+
3537 /**
36- * Get a single block by CID.
38+ * Get a single block by CID
3739 *
3840 * @param {CID } cid
3941 * @param {Object } options
4042 * @returns {Promise<Block> }
4143 */
4244 async get ( cid , options ) {
4345 const key = cidToKey ( cid )
44- let blockData
45- try {
46- blockData = await store . get ( key , options )
47- return new Block ( blockData , cid )
48- } catch ( err ) {
49- if ( err . code === 'ERR_NOT_FOUND' ) {
50- const otherCid = cidToOtherVersion ( cid )
51-
52- if ( ! otherCid ) {
53- throw err
54- }
55-
56- const otherKey = cidToKey ( otherCid )
57- const blockData = await store . get ( otherKey , options )
58- await store . put ( key , blockData )
59- return new Block ( blockData , cid )
60- }
46+ const blockData = await store . get ( key , options )
6147
62- throw err
63- }
48+ return new Block ( blockData , cid )
6449 } ,
50+
6551 /**
66- * Like get, but for more.
52+ * Like get, but for more
6753 *
6854 * @param {AsyncIterator<CID> } cids
6955 * @param {Object } options
@@ -74,8 +60,9 @@ function createBaseStore (store) {
7460 yield this . get ( cid , options )
7561 }
7662 } ,
63+
7764 /**
78- * Write a single block to the store.
65+ * Write a single block to the store
7966 *
8067 * @param {Block } block
8168 * @param {Object } options
@@ -86,59 +73,75 @@ function createBaseStore (store) {
8673 throw new Error ( 'invalid block' )
8774 }
8875
89- const exists = await this . has ( block . cid )
76+ const key = cidToKey ( block . cid )
77+ const exists = await store . has ( key , options )
9078
91- if ( exists ) {
92- return this . get ( block . cid , options )
79+ if ( ! exists ) {
80+ await store . put ( key , block . data , options )
9381 }
9482
95- await store . put ( cidToKey ( block . cid ) , block . data , options )
96-
9783 return block
9884 } ,
9985
10086 /**
101- * Like put, but for more.
87+ * Like put, but for more
10288 *
10389 * @param {AsyncIterable<Block>|Iterable<Block> } blocks
10490 * @param {Object } options
10591 * @returns {AsyncIterable<Block> }
10692 */
10793 async * putMany ( blocks , options ) { // eslint-disable-line require-await
108- yield * pipe (
109- blocks ,
110- ( source ) => {
111- // turn them into a key/value pair
112- return map ( source , ( block ) => {
113- return { key : cidToKey ( block . cid ) , value : block . data }
114- } )
115- } ,
116- ( source ) => {
117- // put them into the datastore
118- return store . putMany ( source , options )
119- } ,
120- ( source ) => {
121- // map the returned key/value back into a block
122- return map ( source , ( { key, value } ) => {
123- return new Block ( value , keyToCid ( key ) )
124- } )
94+ // we cannot simply chain to `store.putMany` because we convert a CID into
95+ // a key based on the multihash only, so we lose the version & codec and
96+ // cannot give the user back the CID they used to create the block, so yield
97+ // to `store.putMany` but return the actual block the user passed in.
98+ //
99+ // nb. we want to use `store.putMany` here so bitswap can control batching
100+ // up block HAVEs to send to the network - if we use multiple `store.put`s
101+ // it will not be able to guess we are about to `store.put` more blocks
102+ const output = pushable ( )
103+
104+ // process.nextTick runs on the microtask queue, setImmediate runs on the next
105+ // event loop iteration so is slower. Use process.nextTick if it is available.
106+ const runner = process && process . nextTick ? process . nextTick : setImmediate
107+
108+ runner ( async ( ) => {
109+ try {
110+ await drain ( store . putMany ( async function * ( ) {
111+ for await ( const block of blocks ) {
112+ const key = cidToKey ( block . cid )
113+ const exists = await store . has ( key , options )
114+
115+ if ( ! exists ) {
116+ yield { key, value : block . data }
117+ }
118+
119+ // there is an assumption here that after the yield has completed
120+ // the underlying datastore has finished writing the block
121+ output . push ( block )
122+ }
123+ } ( ) ) )
124+
125+ output . end ( )
126+ } catch ( err ) {
127+ output . end ( err )
125128 }
126- )
129+ } )
130+
131+ yield * output
127132 } ,
133+
128134 /**
129- * Does the store contain block with this cid ?
135+ * Does the store contain block with this CID ?
130136 *
131137 * @param {CID } cid
132138 * @param {Object } options
133139 * @returns {Promise<bool> }
134140 */
135- async has ( cid , options ) {
136- const exists = await store . has ( cidToKey ( cid ) , options )
137- if ( exists ) return exists
138- const otherCid = cidToOtherVersion ( cid )
139- if ( ! otherCid ) return false
140- return store . has ( cidToKey ( otherCid ) , options )
141+ async has ( cid , options ) { // eslint-disable-line require-await
142+ return store . has ( cidToKey ( cid ) , options )
141143 } ,
144+
142145 /**
143146 * Delete a block from the store
144147 *
@@ -149,6 +152,7 @@ function createBaseStore (store) {
149152 async delete ( cid , options ) { // eslint-disable-line require-await
150153 return store . delete ( cidToKey ( cid ) , options )
151154 } ,
155+
152156 /**
153157 * Delete a block from the store
154158 *
@@ -157,12 +161,9 @@ function createBaseStore (store) {
157161 * @returns {Promise<void> }
158162 */
159163 async * deleteMany ( cids , options ) { // eslint-disable-line require-await
160- yield * store . deleteMany ( ( async function * ( ) {
161- for await ( const cid of cids ) {
162- yield cidToKey ( cid )
163- }
164- } ( ) ) , options )
164+ yield * store . deleteMany ( map ( cids , cid => cidToKey ( cid ) ) , options )
165165 } ,
166+
166167 /**
167168 * Close the store
168169 *
@@ -173,11 +174,3 @@ function createBaseStore (store) {
173174 }
174175 }
175176}
176-
177- function cidToOtherVersion ( cid ) {
178- try {
179- return cid . version === 0 ? cid . toV1 ( ) : cid . toV0 ( )
180- } catch ( err ) {
181- return null
182- }
183- }
0 commit comments