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" },