kredits/ipfs-pinner
kredits/ipfs-pinner
9
1
Fork 0
Code Issues 2 Pull Requests 1 Actions Releases 9 Activity

Upgrade for new kredits contracts and RSK #10

Merged
greg merged 13 commits from chore/upgrade_kredits into master 2022-11-02 17:42:39 +00:00
Conversation 1 Commits 13 Files Changed 6 +145 -63
4 changed files with 145 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 6829a13a57 - Show all commits

66
index.js
View File

@ -12,16 +12,19 @@ const argv = require('yargs')
host: 'localhost', host: 'localhost',
port: '5001', port: '5001',
protocol: 'http', protocol: 'http',
monitor: true, watch: false,
progress: false,
bootstrapNode: `${defaultPeers[0].Addrs[0]}/ipfs/${defaultPeers[0].ID}` bootstrapNode: `${defaultPeers[0].Addrs[0]}/ipfs/${defaultPeers[0].ID}`
}) })
.boolean('monitor') .boolean('watch')
.boolean('progress')
.describe({ .describe({
rpcUrl: 'Web3/EVM node RPC URL; alternative to --network', rpcUrl: 'Web3/EVM node RPC URL; alternative to --network',
host: 'IPFS API host', host: 'IPFS API host',
port: 'IPFS API port', port: 'IPFS API port',
protocol: 'IPFS API protocol', 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' 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') .example('$0 --host localhost', 'Pins all existing IPFS documents to the IPFS API running on localhost and monitors for new events')
@ -33,39 +36,48 @@ const ipfsConfig = {
protocol: argv.protocol protocol: argv.protocol
}; };
console.log(`Using IPFS:`, ipfsConfig); debug(`IPFS node:`, ipfsConfig);
(async () => { (async () => {
try { const kredits = await Kredits.for(
const kredits = await Kredits.for( { rpcUrl: argv.rpcUrl },
{ rpcUrl: argv.rpcUrl }, { ipfsConfig: ipfsConfig }
{ ipfsConfig: ipfsConfig } ).init().catch(e => {
).init(); console.log('Failed to initialize Kredits:');
console.log(e.message);
process.exit(1);
});
// Check the connection to the IPFS client // TODO redesign IPFS wrapper API and do not use an internal attribute
// TODO redesign IPFS wrapper API and do not use an internal attribute const ipfsApi = kredits.ipfs._ipfsAPI;
const ipfsApi = kredits.ipfs._ipfsAPI;
await ipfsApi.id(); 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}`); debug(`Connecting to known IPFS node ${argv.bootstrapNode}`);
await ipfsApi.swarm.connect(argv.bootstrapNode); await ipfsApi.swarm.connect(argv.bootstrapNode);
const ipfsPinner = new IpfsPinner(kredits); const ipfsPinner = new IpfsPinner(kredits, {
progress: argv.progress
});
ipfsPinner.pinAll().then(pins => { await ipfsPinner.pinAll().then(cids => {
console.log(`Pinned ${pins.length} existing documents`); console.log(`\nSuccessfully pinned ${cids.length} documents`)
}); });
ipfsPinner.monitor(pin => { if (argv.watch) {
console.log('\nWatching contract events for new documents...');
ipfsPinner.watch(pin => {
console.log('Pinned a new document:', pin[0]["hash"]); console.log('Pinned a new document:', pin[0]["hash"]);
}); });
} else {
// TODO Add new deployment/DAO/org ID or all contract proxy addresses process.exit(0);
// console.log(`Subscribed to DAO: ${kredits.Kernel.contract.address}`);
} catch(e) {
console.log('Failed to start');
console.log(e);
process.exit(1);
} }
// TODO Add new deployment/DAO/org ID or all contract proxy addresses
// console.log(`Subscribed to DAO: ${kredits.Kernel.contract.address}`);
})(); })();

106
lib/ipfs-pinner.js
View File

@ -1,32 +1,62 @@
const debug = require('debug')('ipfs-pinner'); const debug = require('debug')('ipfs-pinner');
const cliProgress = require('cli-progress');
async function promiseAllInBatches(task, items, batchSize) { function sleep(ms) {
let position = 0; return new Promise(resolve => setTimeout(resolve, ms));
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;
} }
class IpfsPinner { class IpfsPinner {
constructor (kredits, ipfsApi) { constructor (kredits, options={}) {
this.kredits = kredits; 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 () { async pinAll () {
const contributorHashes = await this._pinAllFromContract(this.kredits.Contributor); console.log('Pinning IPFS documents for all known items...\n')
const contributionHashes = await this._pinAllFromContract(this.kredits.Contribution); const cids = [];
const reimbursementHashes = await this._pinAllFromContract(this.kredits.Reimbursement); const promises = [];
const contracts = [
this.kredits.Contributor,
this.kredits.Contribution,
// TODO uncomment once we have data here
// this.kredits.Reimbursement
]
return contributorHashes.concat(contributionHashes) for (const contract of contracts) {
.concat(reimbursementHashes); debug(`Pinning data from ${contract.constructor.name}...`);
const itemCount = await contract.count;
debug(`${contract.constructor.name} 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);
// Avoid console output race condition with progress bars finishing update
if (this.progressBars) await sleep(1000);
return cids;
} }
monitor (callback) { watch (callback) {
this.kredits.Contribution.on('ContributionAdded', (id) => { this.kredits.Contribution.on('ContributionAdded', (id) => {
this.kredits.Contribution.getData(id) this.kredits.Contribution.getData(id)
.then(data => { return this.ipfsApi.pin(data); }) .then(data => { return this.ipfsApi.pin(data); })
@ -44,22 +74,40 @@ class IpfsPinner {
}); });
} }
async _pinAllFromContract (contract) { async _pinAllFromContract (contract, itemCount, progressBar) {
debug(`Pinning data from ${contract.constructor.name}...`); const ipfsApi = this.ipfsApi;
const count = await contract.count; const progressBars = this.progressBars;
debug('Item count:', count); const ids = [...Array(itemCount).keys()].map(i => i+1);
const ids = [...Array(count).keys()].map(i => i+1);
const cids = []; const cids = [];
const batchSize = 20;
let position = 0;
async function loadAndPin (id) { async function loadAndPin(id) {
debug(`Loading ${contract.constructor.name} #${id}`); let cid;
return contract.getData(id).then(data => {
debug(`Pinning ${contract.constructor.name} #${id}`); try {
return this.ipfsApi.pin(data).then(cid => cids.push(cid)); const data = await contract.getData(id);
}); debug(`Loaded ${contract.constructor.name} #${id}`);
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 (progressBars) { progressBar.increment(); }
}
} }
await promiseAllInBatches(loadAndPin.bind(this), ids, 100); while (position < itemCount) {
const batchIds = ids.slice(position, position + batchSize);
await Promise.all(batchIds.map(async id => loadAndPin(id)));
position += batchSize;
}
return cids; return cids;
} }

33
package-lock.json generated
View File

@ -9,7 +9,8 @@
"version": "1.2.0", "version": "1.2.0",
"license": "MIT", "license": "MIT",
"dependencies": { "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", "debug": "^4.3.4",
"yargs": "^17.6.0" "yargs": "^17.6.0"
}, },
@ -720,9 +721,9 @@
"integrity": "sha512-yOTK5WiXFDNAitPByMabE365aEEzFHgSUSgAssbJWt7BZ80HQSVu8XWrQiTbFbCkoIBmXwPP/RoxgXJQVgZTFQ==" "integrity": "sha512-yOTK5WiXFDNAitPByMabE365aEEzFHgSUSgAssbJWt7BZ80HQSVu8XWrQiTbFbCkoIBmXwPP/RoxgXJQVgZTFQ=="
}, },
"node_modules/@kredits/contracts": { "node_modules/@kredits/contracts": {
"version": "7.0.0-beta.0", "version": "7.0.0",
"resolved": "git+https://gitea.kosmos.org/kredits/contracts#6e0ec8741e61b51fb5c9c636da4e8d3610d090ac", "resolved": "https://registry.npmjs.org/@kredits/contracts/-/contracts-7.0.0.tgz",
"license": "MIT", "integrity": "sha512-UITEkP3njFNI2WS7v5ivGE3ruFwdWPWuJZrhBXBEAZbtmr1t/p1K7jkmmjyLDUeKXJ/udMlH6oQMCgh7P/aHNg==",
"dependencies": { "dependencies": {
"@kosmos/schemas": "^3.1.0", "@kosmos/schemas": "^3.1.0",
"ethers": "^5.4.7", "ethers": "^5.4.7",
@ -930,6 +931,17 @@
"cborg": "cli.js" "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": { "node_modules/cliui": {
"version": "8.0.1", "version": "8.0.1",
"resolved": "https://registry.npmjs.org/cliui/-/cliui-8.0.1.tgz", "resolved": "https://registry.npmjs.org/cliui/-/cliui-8.0.1.tgz",
@ -2233,8 +2245,9 @@
"integrity": "sha512-yOTK5WiXFDNAitPByMabE365aEEzFHgSUSgAssbJWt7BZ80HQSVu8XWrQiTbFbCkoIBmXwPP/RoxgXJQVgZTFQ==" "integrity": "sha512-yOTK5WiXFDNAitPByMabE365aEEzFHgSUSgAssbJWt7BZ80HQSVu8XWrQiTbFbCkoIBmXwPP/RoxgXJQVgZTFQ=="
}, },
"@kredits/contracts": { "@kredits/contracts": {
"version": "git+https://gitea.kosmos.org/kredits/contracts#6e0ec8741e61b51fb5c9c636da4e8d3610d090ac", "version": "7.0.0",
"from": "@kredits/contracts@git+https://gitea.kosmos.org/kredits/contracts#6e0ec87", "resolved": "https://registry.npmjs.org/@kredits/contracts/-/contracts-7.0.0.tgz",
"integrity": "sha512-UITEkP3njFNI2WS7v5ivGE3ruFwdWPWuJZrhBXBEAZbtmr1t/p1K7jkmmjyLDUeKXJ/udMlH6oQMCgh7P/aHNg==",
"requires": { "requires": {
"@kosmos/schemas": "^3.1.0", "@kosmos/schemas": "^3.1.0",
"ethers": "^5.4.7", "ethers": "^5.4.7",
@ -2402,6 +2415,14 @@
"resolved": "https://registry.npmjs.org/cborg/-/cborg-1.9.5.tgz", "resolved": "https://registry.npmjs.org/cborg/-/cborg-1.9.5.tgz",
"integrity": "sha512-fLBv8wmqtlXqy1Yu+pHzevAIkW6k2K0ZtMujNzWphLsA34vzzg9BHn+5GmZqOJkSA9V7EMKsWrf6K976c1QMjQ==" "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": { "cliui": {
"version": "8.0.1", "version": "8.0.1",
"resolved": "https://registry.npmjs.org/cliui/-/cliui-8.0.1.tgz", "resolved": "https://registry.npmjs.org/cliui/-/cliui-8.0.1.tgz",

3
package.json
View File

@ -20,7 +20,8 @@
], ],
"license": "MIT", "license": "MIT",
"dependencies": { "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", "debug": "^4.3.4",
"yargs": "^17.6.0" "yargs": "^17.6.0"
}, },