rpc-client/dist/query/wisdom.js

130 lines
4.3 KiB
JavaScript
Raw Normal View History

2022-12-04 07:42:04 +00:00
import { clearTimeout } from "timers";
import b4a from "b4a";
import { flatten, isPromise, validateResponse, validateTimestampedResponse, } from "../util.js";
import RPC from "@lumeweb/rpc";
2022-08-28 06:33:49 +00:00
import { blake2b } from "libskynet";
2022-12-04 07:42:04 +00:00
import { ERR_INVALID_SIGNATURE, ERR_NO_RELAYS } from "../error.js";
import RpcQueryBase from "./base.js";
2022-08-29 03:19:45 +00:00
function flatHash(data) {
const flattenedData = flatten(data).sort();
return Buffer.from(blake2b(Buffer.from(JSON.stringify(flattenedData)))).toString("hex");
}
2022-08-28 06:33:49 +00:00
export default class WisdomRpcQuery extends RpcQueryBase {
2022-12-04 07:42:04 +00:00
static _activeRelay;
static get activeRelay() {
return this._activeRelay;
}
get result() {
return this._promise;
}
async _run() {
await this.setupRelay();
await this.queryRelay();
await this.checkResponse();
}
resolve(data, timeout = false) {
clearTimeout(this._timeoutTimer);
this._timeout = timeout;
this._completed = true;
if (timeout) {
data = {
error: "timeout",
};
}
this._promiseResolve?.(data);
}
async queryRelay() {
let activeRelay = WisdomRpcQuery.activeRelay;
let relays = this.getRelays();
if (!relays.length) {
throw new Error(ERR_NO_RELAYS);
}
return this.queryRpc(activeRelay, {
module: "rpc",
method: "broadcast_request",
data: {
request: this._query,
relays,
},
});
}
async checkResponse() {
if (this._error) {
this.resolve({ error: this._error });
return;
}
if (!validateResponse(WisdomRpcQuery.activeRelay.stream.remotePublicKey, this._response)) {
this.resolve({ error: ERR_INVALID_SIGNATURE });
return;
}
let relays = [];
for (const relay in this._response?.relays) {
const resp = this._response?.relays[relay];
if (validateTimestampedResponse(b4a.from(relay, "hex"), resp)) {
relays.push(resp);
}
}
if (!relays.length) {
this.resolve({ error: ERR_NO_RELAYS });
return;
}
const responseObjects = relays.reduce((output, item) => {
const field = item.signedField || "data";
// @ts-ignore
const hash = flatHash(item[field]);
2022-08-28 06:33:49 +00:00
output[hash] = item?.data;
return output;
}, {});
2022-12-04 07:42:04 +00:00
const responses = relays.reduce((output, item) => {
const field = item.signedField || "data";
// @ts-ignore
const hash = flatHash(item[field]);
2022-08-28 06:33:49 +00:00
output[hash] = output[hash] ?? 0;
output[hash]++;
return output;
}, {});
for (const responseHash in responses) {
2022-12-04 07:42:04 +00:00
if (responses[responseHash] / relays.length >=
2022-08-28 06:33:49 +00:00
this._network.majorityThreshold) {
let response = responseObjects[responseHash];
2022-12-04 07:42:04 +00:00
response = { data: response };
2022-09-22 15:04:38 +00:00
this.resolve(response);
2022-08-28 06:33:49 +00:00
break;
}
}
}
getRelays() {
2022-09-22 13:36:04 +00:00
if (this._network.maxRelays === 0 ||
this._network.relays.length <= this._network.maxRelays) {
return this._network.relays;
}
const list = [];
let available = this._network.relays;
2022-09-22 13:37:54 +00:00
while (list.length <= this._network.maxRelays) {
2022-09-22 13:36:04 +00:00
const item = Math.floor(Math.random() * available.length);
list.push(available[item]);
available.splice(item, 1);
}
return list;
2022-08-28 06:33:49 +00:00
}
2022-12-04 07:42:04 +00:00
async setupRelay() {
let active = WisdomRpcQuery.activeRelay;
let relays = this._network.relays;
if (!active) {
if (!relays.length) {
throw new Error(ERR_NO_RELAYS);
}
let relay = relays[Math.floor(Math.random() * relays.length)];
let socket = this._network.dht.connect(b4a.from(relay, "hex"));
if (isPromise(socket)) {
socket = await socket;
}
await socket.opened;
WisdomRpcQuery._activeRelay = new RPC(socket);
socket.once("close", () => {
WisdomRpcQuery._activeRelay = undefined;
});
}
}
2022-08-28 06:33:49 +00:00
}