diff --git a/.gitignore b/.gitignore
index 9116e04..f186c44 100644
--- a/.gitignore
+++ b/.gitignore
@@ -9,3 +9,4 @@ npm-debug.log
claimTrieCache.json
syncState.json
yarn-error.log
+chainquery-config.json
diff --git a/chainquery-config.json.example b/chainquery-config.json.example
new file mode 100644
index 0000000..253ecdc
--- /dev/null
+++ b/chainquery-config.json.example
@@ -0,0 +1,6 @@
+{
+ "host": "chainquery.lbry.com",
+ "user": "lighthouse",
+ "password": "",
+ "db": "chainquery"
+}
diff --git a/decoder/decoder.py b/decoder/decoder.py
deleted file mode 100644
index 49716ec..0000000
--- a/decoder/decoder.py
+++ /dev/null
@@ -1,67 +0,0 @@
-#!/usr/bin/python
-# -*- coding: utf-8 -*-
-import json, os
-from bitcoinrpc.authproxy import AuthServiceProxy
-from lbryschema.decode import smart_decode
-from flask import Flask, url_for
-app = Flask(__name__)
-
-def get_lbrycrdd_connection_details(wallet_conf):
- settings = {"username": "lbry",
- "password": "lbry",
- "rpc_port": 9245}
- if wallet_conf and os.path.exists(wallet_conf):
- with open(wallet_conf, "r") as conf:
- conf_lines = conf.readlines()
- for l in conf_lines:
- if l.startswith("rpcuser="):
- settings["username"] = l[8:].rstrip('\n')
- if l.startswith("rpcpassword="):
- settings["password"] = l[12:].rstrip('\n')
- if l.startswith("rpcport="):
- settings["rpc_port"] = int(l[8:].rstrip('\n'))
- rpc_user = settings["username"]
- rpc_pass = settings["password"]
- rpc_port = settings["rpc_port"]
- rpc_url = "127.0.0.1"
- return "http://%s:%s@%s:%i" % (rpc_user, rpc_pass, rpc_url, rpc_port)
-
-@app.errorhandler(500)
-def internal_error(error):
-
- return 'error when decoding claims'
-
-
-@app.route('/claim_decode/<txid>/<nout>')
-def api_decode(txid, nout):
- connection_string = get_lbrycrdd_connection_details(os.path.expanduser("~")+"/.lbrycrd/lbrycrd.conf")
- rpc = AuthServiceProxy(connection_string)
- result = rpc.getclaimsfortx(txid)
- claim = None
- for claim_out in result:
- if claim_out['nOut'] == int(nout):
- claim = claim_out
- break
- if claim:
- converted = ''.join([chr(ord(i)) for i in claim['value']])
- decoded = smart_decode(converted)
- claim['value'] = decoded.claim_dict
- return json.dumps(claim)
-
-
-@app.route('/claim_decodeinv/<claimid>')
-def api_decodebyclaim(claimid):
- connection_string = get_lbrycrdd_connection_details(os.path.expanduser("~")+"/.lbrycrd/lbrycrd.conf")
- rpc = AuthServiceProxy(connection_string)
- claim = rpc.getvalueforname(claimid)
- if claim:
- converted = ''.join([chr(ord(i)) for i in claim['value']])
- decoded = smart_decode(converted)
- claim['value'] = decoded.claim_dict
- return json.dumps(claim)
-
-if __name__ == '__main__':
- app.run(host='127.0.0.1')
-
-
-
diff --git a/decoder/requirements.txt b/decoder/requirements.txt
deleted file mode 100644
index 116e61b..0000000
--- a/decoder/requirements.txt
+++ /dev/null
@@ -1,3 +0,0 @@
-git+https://github.com/lbryio/lbryschema.git#egg=lbryschema
-python-bitcoinrpc==0.1
-flask
\ No newline at end of file
diff --git a/package-lock.json b/package-lock.json
index c83f795..dda837f 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -2814,6 +2814,11 @@
"resolved": "https://registry.npmjs.org/big.js/-/big.js-5.2.2.tgz",
"integrity": "sha512-vyL2OymJxmarO8gxMr0mhChsO9QGwhynfuu4+MHTAW6czfq9humCB7rKpUjDd9YUiDPU4mzpyupFSvOClAwbmQ=="
},
+ "bignumber.js": {
+ "version": "7.2.1",
+ "resolved": "https://registry.npmjs.org/bignumber.js/-/bignumber.js-7.2.1.tgz",
+ "integrity": "sha512-S4XzBk5sMB+Rcb/LNcpzXr57VRTxgAvaAEDAl1AwRx27j00hT84O6OkteE7u8UB3NuaaygCRrEpqox4uDOrbdQ=="
+ },
"binary-extensions": {
"version": "2.0.0",
"resolved": "https://registry.npmjs.org/binary-extensions/-/binary-extensions-2.0.0.tgz",
@@ -7122,6 +7127,24 @@
"resolved": "https://registry.npmjs.org/mute-stream/-/mute-stream-0.0.6.tgz",
"integrity": "sha1-SJYrGeFp/R38JAs/HnMXYnu8R9s="
},
+ "mysql": {
+ "version": "2.17.1",
+ "resolved": "https://registry.npmjs.org/mysql/-/mysql-2.17.1.tgz",
+ "integrity": "sha512-7vMqHQ673SAk5C8fOzTG2LpPcf3bNt0oL3sFpxPEEFp1mdlDcrLK0On7z8ZYKaaHrHwNcQ/MTUz7/oobZ2OyyA==",
+ "requires": {
+ "bignumber.js": "7.2.1",
+ "readable-stream": "2.3.6",
+ "safe-buffer": "5.1.2",
+ "sqlstring": "2.3.1"
+ },
+ "dependencies": {
+ "safe-buffer": {
+ "version": "5.1.2",
+ "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.1.2.tgz",
+ "integrity": "sha512-Gd2UZBJDkXlY7GbJxfsE8/nvKkUEU1G38c1siN6QP6a9PT9MmHB8GnpscSmMJSoF8LOIrt8ud/wPtojys4G6+g=="
+ }
+ }
+ },
"nan": {
"version": "2.14.0",
"resolved": "https://registry.npmjs.org/nan/-/nan-2.14.0.tgz",
@@ -10984,6 +11007,11 @@
"resolved": "https://registry.npmjs.org/sprintf-js/-/sprintf-js-1.0.3.tgz",
"integrity": "sha1-BOaSb2YolTVPPdAVIDYzuFcpfiw="
},
+ "sqlstring": {
+ "version": "2.3.1",
+ "resolved": "https://registry.npmjs.org/sqlstring/-/sqlstring-2.3.1.tgz",
+ "integrity": "sha1-R1OT/56RR5rqYtyvDKPRSYOn+0A="
+ },
"sshpk": {
"version": "1.16.1",
"resolved": "https://registry.npmjs.org/sshpk/-/sshpk-1.16.1.tgz",
diff --git a/package.json b/package.json
index b0d4b31..8d12626 100644
--- a/package.json
+++ b/package.json
@@ -56,6 +56,7 @@
"koa-logger": "^2.0.0",
"koa-router": "^7.0.0",
"limited-request-queue": "^3.0.4",
+ "mysql": "^2.17.1",
"node-slack": "^0.0.7",
"oas": "^0.8.15",
"ora": "^1.3.0",
diff --git a/server/controllers/lighthouse.js b/server/controllers/lighthouse.js
index 506e583..97c2262 100644
--- a/server/controllers/lighthouse.js
+++ b/server/controllers/lighthouse.js
@@ -48,7 +48,7 @@ function getResults (input) {
'bool': {
'must': {
'query_string': {
- 'fields': ['channel_id'],
+ 'fields': ['channel_claim_id'],
'query' : getEscapedQuery(input.channel_id.trim()),
},
},
diff --git a/server/index.js b/server/index.js
index a110709..2004396 100644
--- a/server/index.js
+++ b/server/index.js
@@ -6,14 +6,14 @@ import helmet from 'koa-helmet';
import routing from './routes/';
import { port } from './config';
import winston from 'winston';
-import slack from 'node-slack';
+import Slack from 'node-slack';
require('winston-daily-rotate-file');
// Setup logging
winston.remove(winston.transports.Console);
winston.add(winston.transports.Console, { colorize: true, timestamp: true, prettyPrint: true });
-var slackAPIKey = process.env.SLACK_HOOK_URL;
-var mySlack = new slack(slackAPIKey, {});
+const slackAPIKey = process.env.SLACK_HOOK_URL;
+const mySlack = new Slack(slackAPIKey, {});
// Create Koa Application
const app = new Koa();
diff --git a/server/utils/chainquery/index.js b/server/utils/chainquery/index.js
index 74f6820..8e9e485 100644
--- a/server/utils/chainquery/index.js
+++ b/server/utils/chainquery/index.js
@@ -13,25 +13,35 @@ import fs from 'fs';
import fileExists from 'file-exists';
import * as util from './util';
import {logErrorToSlack} from '../../index';
+import mysql from 'mysql';
+import chainqueryConfig from '../../../chainquery-config.json';
-const elasticsearchloglevel = 'info';
+let connection = null;
+
+const esLogLevel = 'info';
const MaxClaimsToProcessPerIteration = 100000;
const BatchSize = 5000;
-const loggerStream = winstonStream(winston, elasticsearchloglevel);
+const loggerStream = winstonStream(winston, esLogLevel);
const eclient = new elasticsearch.Client({
host: 'http://localhost:9200',
log: {
- level : elasticsearchloglevel,
+ level : esLogLevel,
type : 'stream',
stream: loggerStream,
},
});
+
const queue = new ElasticQueue({elastic: eclient});
+queue.on('drain', function () {
+ console.log('elasticsearch queue is drained');
+});
// Check that our syncState file exist.
fileExists(path.join(appRoot.path, 'syncState.json'), (err, exists) => {
- if (err) { throw err }
+ if (err) {
+ throw err;
+ }
if (!exists) {
fs.writeFileSync(path.join(appRoot.path, 'syncState.json'), '{}');
}
@@ -56,16 +66,16 @@ export async function claimSync () {
let lastID = syncState.LastID;
let iteration = 0;
while (!finished) {
- let claimsResponse = await getClaimsSince(syncState.LastSyncTime, lastID, BatchSize);
- let claims = JSON.parse(claimsResponse).data;
+ let claims = await getClaimsSince(syncState.LastSyncTime, lastID, BatchSize);
status.info = 'addingClaimsToElastic';
for (let claim of claims) {
if (claim.value === null) {
console.log(claim);
- await logErrorToSlack('Failed to process claim ' + claim.claimId + ' due to missing value');
+ // await logErrorToSlack('Failed to process claim ' + claim.claimId + ' due to missing value');
+ console.error('Failed to process claim ' + claim.claimId + ' due to missing value');
continue;
}
- claim.value = JSON.parse(claim.value).Claim;
+ claim.value = claim.value.Claim;
if (claim.name && claim.value) {
claim.suggest_name = {
input : '' + claim.name + '',
@@ -83,7 +93,7 @@ export async function claimSync () {
finished = claims.length < BatchSize || (iteration * BatchSize + BatchSize >= MaxClaimsToProcessPerIteration);
iteration++;
}
- deleteBlockedClaims();
+ await deleteBlockedClaims();
// If not finished, store last id to run again later where we left off, otherwise update last sync time.
if (iteration * BatchSize + BatchSize >= MaxClaimsToProcessPerIteration) {
syncState.LastID = lastID;
@@ -95,12 +105,12 @@ export async function claimSync () {
status.info = 'upToDate';
status.syncState = syncState;
await sleep(600000);
- claimSync();
+ await claimSync();
} catch (err) {
await logErrorToSlack(err);
status.err = err;
await sleep(600000);
- claimSync();
+ await claimSync();
}
}
@@ -119,25 +129,21 @@ async function deleteBlockedClaims () {
winston.log('info', '[Importer] Done processing blocked claims!');
}
-async function deleteFromElastic (claimid) {
- return new Promise(async (resolve, reject) => {
- queue.push({
- index: 'claims',
- type : 'claim',
- id : claimid,
- body : {},
- });
+function deleteFromElastic (claimid) {
+ queue.push({
+ index: 'claims',
+ type : 'claim',
+ id : claimid,
+ body : {},
});
}
-async function pushElastic (claim) {
- return new Promise(async (resolve, reject) => {
- queue.push({
- index: 'claims',
- type : 'claim',
- id : claim.claimId,
- body : claim,
- });
+function pushElastic (claim) {
+ queue.push({
+ index: 'claims',
+ type : 'claim',
+ id : claim.claimId,
+ body : claim,
});
}
@@ -153,6 +159,7 @@ function getJSON (path) {
});
});
}
+
function saveJSON (path, obj) {
return new Promise((resolve, reject) => {
jsonfile.writeFile(path, obj, function (err, jsoncontent) {
@@ -183,34 +190,66 @@ function getBlockedOutpoints () {
});
}
+function getChainqueryConnection () {
+ if (connection === null) {
+ connection = mysql.createConnection({
+ host : chainqueryConfig.host,
+ user : chainqueryConfig.user,
+ password: chainqueryConfig.password,
+ database: chainqueryConfig.db,
+ });
+ connection.connect();
+ }
+ return connection;
+}
+
function getClaimsSince (time, lastID, MaxClaimsInCall) {
return new Promise((resolve, reject) => {
- let query = `` +
- `SELECT ` +
- `c.id, ` +
- `c.name,` +
- `p.name as channel,` +
- `p.claim_id as channel_id,` +
- `c.bid_state,` +
- `c.effective_amount,` +
- `COALESCE(p.effective_amount,1) as certificate_amount,` +
- `c.claim_id as claimId,` +
- `c.value_as_json as value ` +
- `FROM claim c ` +
- `LEFT JOIN claim p on p.claim_id = c.publisher_id ` +
- `WHERE c.id >` + lastID + ` ` +
- `AND c.modified_at >='` + time + `' ` +
- `ORDER BY c.id ` +
- `LIMIT ` + MaxClaimsInCall;
+ let query = `SELECT c.id,
+ c.name,
+ p.name as channel,
+ p.claim_id as channel_id,
+ c.bid_state,
+ c.effective_amount,
+ COALESCE(p.effective_amount,1) as certificate_amount,
+ c.claim_id as claimId,
+ c.value_as_json as value
+ FROM claim c LEFT JOIN claim p
+ on p.claim_id = c.publisher_id
+ WHERE c.id >${lastID} AND
+ c.modified_at >='${time}'
+ ORDER BY c.id LIMIT ${MaxClaimsInCall}`;
// Outputs full query to console for copy/paste into chainquery (debugging)
console.log(query);
- rp(`https://chainquery.lbry.com/api/sql?query=` + query)
- .then(function (htmlString) {
- resolve(htmlString);
- })
- .catch(function (err) {
+ getChainqueryConnection().query(query, function (err, results, fields) {
+ if (err) {
+ console.error(err);
logErrorToSlack('[Importer] Error getting updated claims. ' + err);
- reject(err);
- });
+ return reject(err);
+ }
+ let claims = [];
+ for (let i = 0; i < results.length; i++) {
+ let r = results[i];
+ let value = null;
+ try {
+ value = JSON.parse(r.value);
+ } catch (e) {
+ console.error(e);
+ console.error(r.value);
+ }
+ claims.push({
+ id : r.id,
+ name : r.name,
+ channel : r.channel,
+ channel_claim_id : r.channel_id,
+ bid_state : r.bid_state,
+ effective_amount : r.effective_amount,
+ certificate_amount: r.certificate_amount,
+ claimId : r.claimId,
+ value : value,
+ });
+ }
+ resolve(claims);
+ });
});
}