From 0ac7228bfe7a9b1eac33ca740a3f6c82e95cbd74 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A2u=20Cao?= Date: Tue, 1 Nov 2022 22:10:57 +0100 Subject: [PATCH 1/6] Add CLI option for progress bars Also refactors some of the other output and options. --- index.js | 31 ++++++++++------ lib/ipfs-pinner.js | 89 +++++++++++++++++++++++++++++----------------- package-lock.json | 20 +++++++++++ package.json | 1 + 4 files changed, 99 insertions(+), 42 deletions(-) diff --git a/index.js b/index.js index c528e50..1751695 100755 --- a/index.js +++ b/index.js @@ -12,16 +12,19 @@ const argv = require('yargs') host: 'localhost', port: '5001', protocol: 'http', - monitor: true, + watch: true, + progress: false, bootstrapNode: `${defaultPeers[0].Addrs[0]}/ipfs/${defaultPeers[0].ID}` }) - .boolean('monitor') + .boolean('watch') + .boolean('progress') .describe({ rpcUrl: 'Web3/EVM node RPC URL; alternative to --network', host: 'IPFS API host', port: 'IPFS API port', protocol: 'IPFS API protocol', - monitor: 'Monitor contract events for new IPFS documents', + watch: 'Monitor contract events for new IPFS documents', + progress: 'Show progress bars', bootstrapNode: 'IPFS node address to connect to before fetching documents' }) .example('$0 --host localhost', 'Pins all existing IPFS documents to the IPFS API running on localhost and monitors for new events') @@ -33,7 +36,7 @@ const ipfsConfig = { protocol: argv.protocol }; -console.log(`Using IPFS:`, ipfsConfig); +debug(`IPFS node:`, ipfsConfig); (async () => { try { @@ -51,16 +54,24 @@ console.log(`Using IPFS:`, ipfsConfig); debug(`Connecting to known IPFS node ${argv.bootstrapNode}`); await ipfsApi.swarm.connect(argv.bootstrapNode); - const ipfsPinner = new IpfsPinner(kredits); - - ipfsPinner.pinAll().then(pins => { - console.log(`Pinned ${pins.length} existing documents`); + const ipfsPinner = new IpfsPinner(kredits, { + progress: argv.progress }); - ipfsPinner.monitor(pin => { - console.log('Pinned a new document:', pin[0]["hash"]); + await ipfsPinner.pinAll().then(cids => { + console.log(`\nSuccessfully pinned ${cids.length} documents`) }); + if (argv.watch) { + console.log('\nWatching contract events for new documents...'); + + ipfsPinner.watch(pin => { + console.log('Pinned a new document:', pin[0]["hash"]); + }); + } else { + process.exit(0); + } + // TODO Add new deployment/DAO/org ID or all contract proxy addresses // console.log(`Subscribed to DAO: ${kredits.Kernel.contract.address}`); } catch(e) { diff --git a/lib/ipfs-pinner.js b/lib/ipfs-pinner.js index fdf34f1..284e20e 100644 --- a/lib/ipfs-pinner.js +++ b/lib/ipfs-pinner.js @@ -1,32 +1,55 @@ const debug = require('debug')('ipfs-pinner'); - -async function promiseAllInBatches(task, items, batchSize) { - let position = 0; - let results = []; - while (position < items.length) { - const itemsForBatch = items.slice(position, position + batchSize); - results = [...results, ...await Promise.allSettled(itemsForBatch.map(item => task(item)))]; - position += batchSize; - } - return results; -} +const cliProgress = require('cli-progress'); class IpfsPinner { - constructor (kredits, ipfsApi) { + constructor (kredits, options={}) { this.kredits = kredits; - this.ipfsApi = ipfsApi || this.kredits.ipfs; + this.ipfsApi = this.kredits.ipfs; + this.progressBars = !!options.progress && !process.env.DEBUG; + + if (this.progressBars) { + this.multibar = new cliProgress.MultiBar({ + stopOnComplete: true, + clearOnComplete: false, + hideCursor: false, + etaBuffer: 30, + format: '{entity} [{bar}] {percentage}% | ETA: {eta_formatted} | {value}/{total}' + }, cliProgress.Presets.shades_grey); + } } async pinAll () { - const contributorHashes = await this._pinAllFromContract(this.kredits.Contributor); - const contributionHashes = await this._pinAllFromContract(this.kredits.Contribution); - const reimbursementHashes = await this._pinAllFromContract(this.kredits.Reimbursement); + console.log('Pinning IPFS documents for all known items...\n') + const cids = []; + const promises = []; + const contracts = [ + this.kredits.Contributor, + this.kredits.Contribution, + // TODO uncomment once we have data here + // this.kredits.Reimbursement + ] - return contributorHashes.concat(contributionHashes) - .concat(reimbursementHashes); + for (const contract of contracts) { + debug(`Pinning data from ${contract.constructor.name}...`); + const itemCount = await contract.count; + debug('Item count:', itemCount); + let bar; + + if (this.progressBars) { + bar = this.multibar.create(itemCount, 0); + bar.update(0, {entity: `${contract.constructor.name}s`.padEnd(14)}); + } + + promises.push(this._pinAllFromContract(contract, itemCount, bar) + .then(res => { cids.push(...res); })); + } + + await Promise.all(promises); + + return cids; } - monitor (callback) { + watch (callback) { this.kredits.Contribution.on('ContributionAdded', (id) => { this.kredits.Contribution.getData(id) .then(data => { return this.ipfsApi.pin(data); }) @@ -44,23 +67,25 @@ class IpfsPinner { }); } - async _pinAllFromContract (contract) { - debug(`Pinning data from ${contract.constructor.name}...`); - const count = await contract.count; - debug('Item count:', count); - const ids = [...Array(count).keys()].map(i => i+1); + async _pinAllFromContract (contract, itemCount, progressBar) { + const ids = [...Array(itemCount).keys()].map(i => i+1); const cids = []; + const batchSize = 20; + let position = 0; - async function loadAndPin (id) { - debug(`Loading ${contract.constructor.name} #${id}`); - return contract.getData(id).then(data => { - debug(`Pinning ${contract.constructor.name} #${id}`); - return this.ipfsApi.pin(data).then(cid => cids.push(cid)); - }); + while (position < itemCount) { + const batchIds = ids.slice(position, position + batchSize); + await Promise.all(batchIds.map(async id => { + const data = await contract.getData(id); + debug(`Loaded ${contract.constructor.name} #${id}`); + const cid = await this.ipfsApi.pin(data); + debug(`Pinned ${contract.constructor.name} #${id} at ${cid}`); + cids.push(cid); + if (this.progressBars) { progressBar.increment(); } + })); + position += batchSize; } - await promiseAllInBatches(loadAndPin.bind(this), ids, 100); - return cids; } } diff --git a/package-lock.json b/package-lock.json index 856dba2..ca27a07 100644 --- a/package-lock.json +++ b/package-lock.json @@ -10,6 +10,7 @@ "license": "MIT", "dependencies": { "@kredits/contracts": "git+https://gitea.kosmos.org/kredits/contracts#6e0ec87", + "cli-progress": "^3.11.2", "debug": "^4.3.4", "yargs": "^17.6.0" }, @@ -930,6 +931,17 @@ "cborg": "cli.js" } }, + "node_modules/cli-progress": { + "version": "3.11.2", + "resolved": "https://registry.npmjs.org/cli-progress/-/cli-progress-3.11.2.tgz", + "integrity": "sha512-lCPoS6ncgX4+rJu5bS3F/iCz17kZ9MPZ6dpuTtI0KXKABkhyXIdYB3Inby1OpaGti3YlI3EeEkM9AuWpelJrVA==", + "dependencies": { + "string-width": "^4.2.3" + }, + "engines": { + "node": ">=4" + } + }, "node_modules/cliui": { "version": "8.0.1", "resolved": "https://registry.npmjs.org/cliui/-/cliui-8.0.1.tgz", @@ -2402,6 +2414,14 @@ "resolved": "https://registry.npmjs.org/cborg/-/cborg-1.9.5.tgz", "integrity": "sha512-fLBv8wmqtlXqy1Yu+pHzevAIkW6k2K0ZtMujNzWphLsA34vzzg9BHn+5GmZqOJkSA9V7EMKsWrf6K976c1QMjQ==" }, + "cli-progress": { + "version": "3.11.2", + "resolved": "https://registry.npmjs.org/cli-progress/-/cli-progress-3.11.2.tgz", + "integrity": "sha512-lCPoS6ncgX4+rJu5bS3F/iCz17kZ9MPZ6dpuTtI0KXKABkhyXIdYB3Inby1OpaGti3YlI3EeEkM9AuWpelJrVA==", + "requires": { + "string-width": "^4.2.3" + } + }, "cliui": { "version": "8.0.1", "resolved": "https://registry.npmjs.org/cliui/-/cliui-8.0.1.tgz", diff --git a/package.json b/package.json index 7dc75cb..76b6bf0 100644 --- a/package.json +++ b/package.json @@ -21,6 +21,7 @@ "license": "MIT", "dependencies": { "@kredits/contracts": "git+https://gitea.kosmos.org/kredits/contracts#6e0ec87", + "cli-progress": "^3.11.2", "debug": "^4.3.4", "yargs": "^17.6.0" }, -- 2.25.1 From 4888d1be7892b65ac72340891ede194591ae07fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A2u=20Cao?= Date: Wed, 2 Nov 2022 12:57:53 +0100 Subject: [PATCH 2/6] Don't automatically watch for events --- index.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/index.js b/index.js index 1751695..077f6d4 100755 --- a/index.js +++ b/index.js @@ -12,7 +12,7 @@ const argv = require('yargs') host: 'localhost', port: '5001', protocol: 'http', - watch: true, + watch: false, progress: false, bootstrapNode: `${defaultPeers[0].Addrs[0]}/ipfs/${defaultPeers[0].ID}` }) -- 2.25.1 From c21512fa0b50c65e7a9f353b78c1db427af5fe16 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A2u=20Cao?= Date: Wed, 2 Nov 2022 12:58:07 +0100 Subject: [PATCH 3/6] Add error handling and retries for loading/pinning --- lib/ipfs-pinner.js | 30 +++++++++++++++++++++++------- 1 file changed, 23 insertions(+), 7 deletions(-) diff --git a/lib/ipfs-pinner.js b/lib/ipfs-pinner.js index 284e20e..16869b0 100644 --- a/lib/ipfs-pinner.js +++ b/lib/ipfs-pinner.js @@ -32,7 +32,7 @@ class IpfsPinner { for (const contract of contracts) { debug(`Pinning data from ${contract.constructor.name}...`); const itemCount = await contract.count; - debug('Item count:', itemCount); + debug(`${contract.constructor.name} item count:`, itemCount); let bar; if (this.progressBars) { @@ -68,21 +68,37 @@ class IpfsPinner { } async _pinAllFromContract (contract, itemCount, progressBar) { + const ipfsApi = this.ipfsApi; + const progressBars = this.progressBars; const ids = [...Array(itemCount).keys()].map(i => i+1); const cids = []; const batchSize = 20; let position = 0; - while (position < itemCount) { - const batchIds = ids.slice(position, position + batchSize); - await Promise.all(batchIds.map(async id => { + async function loadAndPin(id) { + let cid; + + try { const data = await contract.getData(id); debug(`Loaded ${contract.constructor.name} #${id}`); - const cid = await this.ipfsApi.pin(data); + cid = await ipfsApi.pin(data); debug(`Pinned ${contract.constructor.name} #${id} at ${cid}`); + } catch(e) { + debug(`Error while trying to load an pin ${contract.constructor.name} #${id}:`) + debug(e); + debug(`\nTrying again...`); + loadAndPin(id); + } finally { cids.push(cid); - if (this.progressBars) { progressBar.increment(); } - })); + if (progressBars) { progressBar.increment(); } + } + } + + while (position < itemCount) { + const batchIds = ids.slice(position, position + batchSize); + + await Promise.all(batchIds.map(async id => loadAndPin(id))); + position += batchSize; } -- 2.25.1 From 9dffb15f45cfb021d67e451ca9c47ed187df04c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A2u=20Cao?= Date: Wed, 2 Nov 2022 13:27:55 +0100 Subject: [PATCH 4/6] Improve initialization --- index.js | 77 ++++++++++++++++++++++++++++---------------------------- 1 file changed, 39 insertions(+), 38 deletions(-) diff --git a/index.js b/index.js index 077f6d4..5e89146 100755 --- a/index.js +++ b/index.js @@ -39,44 +39,45 @@ const ipfsConfig = { debug(`IPFS node:`, ipfsConfig); (async () => { - try { - const kredits = await Kredits.for( - { rpcUrl: argv.rpcUrl }, - { ipfsConfig: ipfsConfig } - ).init(); - - // Check the connection to the IPFS client - // TODO redesign IPFS wrapper API and do not use an internal attribute - const ipfsApi = kredits.ipfs._ipfsAPI; - - await ipfsApi.id(); - - debug(`Connecting to known IPFS node ${argv.bootstrapNode}`); - await ipfsApi.swarm.connect(argv.bootstrapNode); - - const ipfsPinner = new IpfsPinner(kredits, { - progress: argv.progress - }); - - await ipfsPinner.pinAll().then(cids => { - console.log(`\nSuccessfully pinned ${cids.length} documents`) - }); - - if (argv.watch) { - console.log('\nWatching contract events for new documents...'); - - ipfsPinner.watch(pin => { - console.log('Pinned a new document:', pin[0]["hash"]); - }); - } else { - process.exit(0); - } - - // TODO Add new deployment/DAO/org ID or all contract proxy addresses - // console.log(`Subscribed to DAO: ${kredits.Kernel.contract.address}`); - } catch(e) { - console.log('Failed to start'); - console.log(e); + const kredits = await Kredits.for( + { rpcUrl: argv.rpcUrl }, + { ipfsConfig: ipfsConfig } + ).init().catch(e => { + console.log('Failed to initialize Kredits:'); + console.log(e.message); process.exit(1); + }); + + // TODO redesign IPFS wrapper API and do not use an internal attribute + const ipfsApi = kredits.ipfs._ipfsAPI; + + await ipfsApi.id().catch(e => { + console.log('Failed to initialize IPFS:'); + console.log(e.message); + process.exit(1); + }); + + debug(`Connecting to known IPFS node ${argv.bootstrapNode}`); + await ipfsApi.swarm.connect(argv.bootstrapNode); + + const ipfsPinner = new IpfsPinner(kredits, { + progress: argv.progress + }); + + await ipfsPinner.pinAll().then(cids => { + console.log(`\nSuccessfully pinned ${cids.length} documents`) + }); + + if (argv.watch) { + console.log('\nWatching contract events for new documents...'); + + ipfsPinner.watch(pin => { + console.log('Pinned a new document:', pin[0]["hash"]); + }); + } else { + process.exit(0); } + + // TODO Add new deployment/DAO/org ID or all contract proxy addresses + // console.log(`Subscribed to DAO: ${kredits.Kernel.contract.address}`); })(); -- 2.25.1 From 03ee17f00ed285d6b46c54556e77a39cfe09169b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A2u=20Cao?= Date: Wed, 2 Nov 2022 13:34:51 +0100 Subject: [PATCH 5/6] Fix progress bar rendering when finished --- lib/ipfs-pinner.js | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/lib/ipfs-pinner.js b/lib/ipfs-pinner.js index 16869b0..9e0aebc 100644 --- a/lib/ipfs-pinner.js +++ b/lib/ipfs-pinner.js @@ -1,6 +1,10 @@ const debug = require('debug')('ipfs-pinner'); const cliProgress = require('cli-progress'); +function sleep(ms) { + return new Promise(resolve => setTimeout(resolve, ms)); +} + class IpfsPinner { constructor (kredits, options={}) { this.kredits = kredits; @@ -46,6 +50,9 @@ class IpfsPinner { await Promise.all(promises); + // Avoid console output race condition with progress bars finishing update + if (this.progressBars) await sleep(1000); + return cids; } -- 2.25.1 From 328905814d7122f6988b72179a42d7ef48c6b569 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?R=C3=A2u=20Cao?= Date: Wed, 2 Nov 2022 18:27:59 +0100 Subject: [PATCH 6/6] Use new @kredits/contracts release --- package-lock.json | 13 +++++++------ package.json | 2 +- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/package-lock.json b/package-lock.json index ca27a07..cc171c0 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,7 +9,7 @@ "version": "1.2.0", "license": "MIT", "dependencies": { - "@kredits/contracts": "git+https://gitea.kosmos.org/kredits/contracts#6e0ec87", + "@kredits/contracts": "7.0.0", "cli-progress": "^3.11.2", "debug": "^4.3.4", "yargs": "^17.6.0" @@ -721,9 +721,9 @@ "integrity": "sha512-yOTK5WiXFDNAitPByMabE365aEEzFHgSUSgAssbJWt7BZ80HQSVu8XWrQiTbFbCkoIBmXwPP/RoxgXJQVgZTFQ==" }, "node_modules/@kredits/contracts": { - "version": "7.0.0-beta.0", - "resolved": "git+https://gitea.kosmos.org/kredits/contracts#6e0ec8741e61b51fb5c9c636da4e8d3610d090ac", - "license": "MIT", + "version": "7.0.0", + "resolved": "https://registry.npmjs.org/@kredits/contracts/-/contracts-7.0.0.tgz", + "integrity": "sha512-UITEkP3njFNI2WS7v5ivGE3ruFwdWPWuJZrhBXBEAZbtmr1t/p1K7jkmmjyLDUeKXJ/udMlH6oQMCgh7P/aHNg==", "dependencies": { "@kosmos/schemas": "^3.1.0", "ethers": "^5.4.7", @@ -2245,8 +2245,9 @@ "integrity": "sha512-yOTK5WiXFDNAitPByMabE365aEEzFHgSUSgAssbJWt7BZ80HQSVu8XWrQiTbFbCkoIBmXwPP/RoxgXJQVgZTFQ==" }, "@kredits/contracts": { - "version": "git+https://gitea.kosmos.org/kredits/contracts#6e0ec8741e61b51fb5c9c636da4e8d3610d090ac", - "from": "@kredits/contracts@git+https://gitea.kosmos.org/kredits/contracts#6e0ec87", + "version": "7.0.0", + "resolved": "https://registry.npmjs.org/@kredits/contracts/-/contracts-7.0.0.tgz", + "integrity": "sha512-UITEkP3njFNI2WS7v5ivGE3ruFwdWPWuJZrhBXBEAZbtmr1t/p1K7jkmmjyLDUeKXJ/udMlH6oQMCgh7P/aHNg==", "requires": { "@kosmos/schemas": "^3.1.0", "ethers": "^5.4.7", diff --git a/package.json b/package.json index 76b6bf0..3f30c8f 100644 --- a/package.json +++ b/package.json @@ -20,7 +20,7 @@ ], "license": "MIT", "dependencies": { - "@kredits/contracts": "git+https://gitea.kosmos.org/kredits/contracts#6e0ec87", + "@kredits/contracts": "7.0.0", "cli-progress": "^3.11.2", "debug": "^4.3.4", "yargs": "^17.6.0" -- 2.25.1