Added synctoolv2, claimTrie resolving instead of full chain!

Added synctoolv2, claimTrie resolving instead of full chain and some small patches!
This commit is contained in:
Fillerino 2017-09-03 14:47:40 +02:00
parent 1d9e54458f
commit 5c929f343c
6 changed files with 143 additions and 113 deletions

1
claimTrieCache.json Normal file
View file

@ -0,0 +1 @@
[]

View file

@ -23,7 +23,7 @@
"homepage": "https://github.com/lbryio/lighthouse#readme",
"main": "server/index.js",
"scripts": {
"start": "nodemon server/ --exec babel-node",
"start": "nodemon server/ --exec babel-node --ignore 'claimTrieCache.json'",
"build": "babel server -d dist",
"lint": "eslint ./server",
"test": "npm run lint && npm run mocha",
@ -31,10 +31,12 @@
"mocha": "./node_modules/.bin/mocha --compilers js:babel-register --require babel-polyfill"
},
"dependencies": {
"app-root-path": "^2.0.1",
"babel-polyfill": "^6.5.0",
"bitcoin-promise": "filipnyquist/node-bitcoin-promise#c3eb4bea552a7a136a4a1405d831da3e92f2efea",
"bitcoin-promise": "filipnyquist/node-bitcoin-promise#1fbf1cb8913ca3542b66060d48ebea185661e0a7",
"bluebird": "^3.5.0",
"chalk": "^2.0.1",
"elastic-queue": "^0.3.0",
"elasticsearch": "^13.2.0",
"glob": "^7.1.1",
"jsonfile": "^3.0.1",

View file

@ -34,7 +34,7 @@ class LighthouseControllers {
// Start syncing blocks...
startSync () {
winston.log('info', '[Importer] Started importer, indexing claims.');
sync(200000);
sync();
}
/**
* Search api here

View file

@ -1,60 +0,0 @@
'use strict';
const Promise = require('bluebird');
const rp = require('request-promise');
let client;
async function getClaims (height, gclient) {
return new Promise(async (resolve, reject) => {
try {
client = gclient;
let blockHash = await client.getBlockHash(height).then(blockHash => { return blockHash }).catch(err => reject(err));
let block = await client.getBlock(blockHash).then(block => { return block }).catch(err => reject(err));
let claims = await getClaimsForTxes(block.tx, height); // should return an array of claims, decoded if possible.
resolve(claims);
} catch (err) {
return reject(err);
}
});
}
async function getClaimsForTxes (txes, height) {
return new Promise(async (resolve, reject) => {
try {
let claimsArr = [];
for (let tx of txes) {
let claimsTx = await client.getClaimsForTx(tx).then(claims => { return claims }).catch(err => reject(err));
if (claimsTx != null) {
for (let claim of claimsTx) {
claim['height'] = height;
let dClaim = await getValue(tx, claim['nOut']);
if (dClaim !== 'error when decoding claims' && claim['value']) {
claim['value'] = JSON.parse(dClaim);
claimsArr.push(claim);
} else {
claim['value'] = { error: 'non_decodable' };
claimsArr.push(claim);
}
}
}
}
resolve(claimsArr);
} catch (err) {
return reject(err);
}
});
}
async function getValue (tx, i) {
return new Promise(async (resolve, reject) => {
rp(`http://localhost:5000/claim_decode/${tx}/${i}`)
.then(function (htmlString) {
resolve(htmlString);
})
.catch(function (err) {
reject(err);
});
});
}
module.exports = exports = getClaims;

View file

@ -1,13 +1,17 @@
/*
* Importer code, handles all the syncing with the blockchain into elasticsearch.
*/
const bitcoin = require('bitcoin-promise');
const elasticsearch = require('elasticsearch');
const winston = require('winston');
const winstonStream = require('winston-stream');
import bitcoin from 'bitcoin-promise';
import elasticsearch from 'elasticsearch';
import ElasticQueue from 'elastic-queue';
import winston from 'winston';
import winstonStream from 'winston-stream';
import jsonfile from 'jsonfile';
import path from 'path';
import rp from 'request-promise';
import appRoot from 'app-root-path';
const loggerStream = winstonStream(winston, 'info');
const eclient = new elasticsearch.Client({
host: 'http://elastic:changeme@localhost:9200',
log : {
@ -16,7 +20,7 @@ const eclient = new elasticsearch.Client({
stream: loggerStream,
},
});
const queue = new ElasticQueue({elastic: eclient});
const client = new bitcoin.Client({
host : 'localhost',
port : 9245,
@ -24,26 +28,64 @@ const client = new bitcoin.Client({
pass : 'lbry',
timeout: 30000,
});
let claimsSynced = 0;
let status = {};
export async function sync (currentHeight) {
export async function sync () {
try {
let maxHeight = await client.getBlockCount().then(blockHash => { return blockHash }).catch(err => { throw err });
if (currentHeight <= maxHeight) {
let claims = await require('./getClaims')(currentHeight, client);
send(claims);
claimsSynced += claims.length;
// currentHeight / maxHeight / claimsSynced
status.message = `Running,${currentHeight} / ${maxHeight} done, ${claimsSynced} claims imported.`;
sync(currentHeight + 1);
} else {
await sleep(2000);
status.message = `All claims imported, waiting for new blocks at ${maxHeight}`;
sync(currentHeight); // eslint-disable-line no-unreachable
status.info = 'Grabbing the claimTrie...';
let claimTrie = await client.getClaimsInTrie().then(claimtrie => { return claimtrie }).catch(err => { throw err });
let txList = [];
let latestClaimTrie = [];
for (let i in claimTrie) {
for (let o in claimTrie[i].claims) {
txList.push({
txid : claimTrie[i].claims[o].txid,
nOut : claimTrie[i].claims[o].n,
claimId: claimTrie[i].claims[o].claimId,
});
latestClaimTrie.push(claimTrie[i].claims[o].claimId);
}
}
let oldClaimTrie = await getJSON(path.join(appRoot.path, 'claimTrieCache.json')); // get our old claimTrieCache....
let added = await getAddedClaims(oldClaimTrie, latestClaimTrie); // get all new that should be added
let removed = await getRemovedClaims(oldClaimTrie, latestClaimTrie); // get all old that should be removed
status.info = 'Adding/Removing Claims, please wait...';
for (let claimId of added) { // for all new get their tx info and add to database
let tx = txList.find(x => x.claimId === claimId);
if (typeof tx !== 'undefined') {
let value = await getValue(tx.txid, tx.nOut);
if ((value !== 'error when decoding claims')) {
value = JSON.parse(value);
if (value['is controlling'] && value['name'] !== '') {
if (value.name && value.value) {
value.suggest_name = {
input : value.name,
weight: 30,
};
}
pushElastic(value);
}
}
}
}
for (let claimId of removed) { // Call elastic and remove claim by id if it exists.
eclient.delete({
index: 'claims',
type : 'claim',
id : claimId,
}, function (error, response) {
if (error) {
winston.log(error);
}
});
}
// Done adding, update our claimTrie cache to latest and wait a bit...
await saveJSON(path.join(appRoot.path, 'claimTrieCache.json'), latestClaimTrie);
status.info = 'Done updating the claimTrieCache, waiting 5 minutes before doing a recheck..';
await sleep(300000);
sync();
} catch (err) {
console.log(err);
winston.log(err);
status.err = err;
}
}
@ -52,31 +94,64 @@ export function getStats () {
return status;
}
function send (arr) { // Modular change output here :)
arr.forEach(function (claim) {
claim['id'] = claim['claimId'];
// Check if our value is a object, else make it a object...
claim['value'] = (typeof claim.value === 'object' ? claim.value : JSON.parse(claim.value));
// claim['value'] = JSON.stringify(claim['value']);
if (claim.name && claim.value) {
claim.suggest_name = {
input : claim.name,
weight: 30,
};
if (claim.value.claimType === 'streamType' && claim.value.stream.metadata && claim.value.stream.metadata.description) {
claim.suggest_desc = {
input : claim.value.stream.metadata.description.split(' '),
weight: 10,
};
}
}
eclient.create({
function getAddedClaims (oldClaimTrie, newClaimTrie) {
return new Promise((resolve, reject) => {
let a = new Set(oldClaimTrie);
let b = new Set(newClaimTrie);
resolve(new Set([...b].filter(x => !a.has(x))));
});
}
function getRemovedClaims (oldClaimTrie, newClaimTrie) {
return new Promise((resolve, reject) => {
let a = new Set(oldClaimTrie);
let b = new Set(newClaimTrie);
resolve(new Set([...a].filter(x => !b.has(x))));
});
}
function getValue (tx, i) {
return new Promise((resolve, reject) => {
rp(`http://localhost:5000/claim_decode/${tx}/${i}`)
.then(function (htmlString) {
resolve(htmlString);
})
.catch(function (err) {
reject(err);
});
});
}
async function pushElastic (claim) {
return new Promise(async(resolve, reject) => {
queue.push({
index: 'claims',
type : 'claim',
id : claim.claimId,
body : claim,
}, function (error, response) {
if (error) { status.err = error; console.log(error) }
});
});
}
function getJSON (path) {
return new Promise((resolve, reject) => {
jsonfile.readFile(path, function (err, jsoncontent) {
if (err) {
reject(err);
} else {
resolve(jsoncontent);
}
});
});
}
function saveJSON (path, obj) {
return new Promise((resolve, reject) => {
jsonfile.writeFile(path, obj, function (err, jsoncontent) {
if (err) {
reject(err);
} else {
resolve();
}
});
});
}

View file

@ -75,6 +75,10 @@ anymatch@^1.3.0:
micromatch "^2.1.5"
normalize-path "^2.0.0"
app-root-path@^2.0.1:
version "2.0.1"
resolved "https://registry.yarnpkg.com/app-root-path/-/app-root-path-2.0.1.tgz#cd62dcf8e4fd5a417efc664d2e5b10653c651b46"
aproba@^1.0.3:
version "1.1.2"
resolved "https://registry.yarnpkg.com/aproba/-/aproba-1.1.2.tgz#45c6629094de4e96f693ef7eab74ae079c240fc1"
@ -140,7 +144,7 @@ async-each@^1.0.0:
version "1.0.1"
resolved "https://registry.yarnpkg.com/async-each/-/async-each-1.0.1.tgz#19d386a1d9edc6e7c1c85d388aedbcc56d33602d"
async@^1.5.2:
"async@>=0.9 <2.0", async@^1.5.2:
version "1.5.2"
resolved "https://registry.yarnpkg.com/async/-/async-1.5.2.tgz#ec6a61ae56480c0c3cb241c95618e20892f9672a"
@ -678,15 +682,15 @@ binary-extensions@^1.0.0:
version "1.10.0"
resolved "https://registry.yarnpkg.com/binary-extensions/-/binary-extensions-1.10.0.tgz#9aeb9a6c5e88638aad171e167f5900abe24835d0"
bitcoin-promise@filipnyquist/node-bitcoin-promise#c3eb4bea552a7a136a4a1405d831da3e92f2efea:
bitcoin-promise@filipnyquist/node-bitcoin-promise#1fbf1cb8913ca3542b66060d48ebea185661e0a7:
version "1.3.2"
resolved "https://codeload.github.com/filipnyquist/node-bitcoin-promise/tar.gz/c3eb4bea552a7a136a4a1405d831da3e92f2efea"
resolved "https://codeload.github.com/filipnyquist/node-bitcoin-promise/tar.gz/1fbf1cb8913ca3542b66060d48ebea185661e0a7"
dependencies:
bitcoin filipnyquist/node-bitcoin
bitcoin filipnyquist/node-bitcoin#17552efad852a0ae929dc153988649259536a23d
bitcoin@filipnyquist/node-bitcoin:
bitcoin@filipnyquist/node-bitcoin#17552efad852a0ae929dc153988649259536a23d:
version "3.0.2"
resolved "https://codeload.github.com/filipnyquist/node-bitcoin/tar.gz/6e39c26a6b462ad315832bfc8f9e6411e1d8e1aa"
resolved "https://codeload.github.com/filipnyquist/node-bitcoin/tar.gz/17552efad852a0ae929dc153988649259536a23d"
block-stream@*:
version "0.0.9"
@ -1125,7 +1129,15 @@ ee-first@1.1.1:
version "1.1.1"
resolved "https://registry.yarnpkg.com/ee-first/-/ee-first-1.1.1.tgz#590c61156b0ae2f4f0255732a158b266bc56b21d"
elasticsearch@^13.2.0:
elastic-queue@^0.3.0:
version "0.3.0"
resolved "https://registry.yarnpkg.com/elastic-queue/-/elastic-queue-0.3.0.tgz#36fbd56543a5e1c10fff118e09f69293d41ac2fd"
dependencies:
async ">=0.9 <2.0"
elasticsearch ">4.0"
hoek "^2.11.1"
elasticsearch@>4.0, elasticsearch@^13.2.0:
version "13.3.1"
resolved "https://registry.yarnpkg.com/elasticsearch/-/elasticsearch-13.3.1.tgz#c530aea9afb17ea91c3d0a56f1f111ba49bc9239"
dependencies:
@ -1729,7 +1741,7 @@ hide-powered-by@1.0.0:
version "1.0.0"
resolved "https://registry.yarnpkg.com/hide-powered-by/-/hide-powered-by-1.0.0.tgz#4a85ad65881f62857fc70af7174a1184dccce32b"
hoek@2.x.x:
hoek@2.x.x, hoek@^2.11.1:
version "2.16.3"
resolved "https://registry.yarnpkg.com/hoek/-/hoek-2.16.3.tgz#20bb7403d3cea398e91dc4710a8ff1b8274a25ed"