LumeWeb
/
dht-data
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: LumeWeb:78f3edd71d514f9b13bce413b6925d0c7123c6cf
Branches Tags
LumeWeb:master
..
compare: LumeWeb:579bc2bf5df2df7a8f5799bf3c1248600e927394
Branches Tags
LumeWeb:master

No commits in common. "78f3edd71d514f9b13bce413b6925d0c7123c6cf" and "579bc2bf5df2df7a8f5799bf3c1248600e927394" have entirely different histories.
78f3edd71d ... 579bc2bf5d

15 changed files with 112 additions and 306 deletions
Show Stats Expand all files Collapse all files

1
dist/DHTDataBase.d.ts.map vendored
View File

@ -1 +0,0 @@
{"version":3,"file":"DHTDataBase.d.ts","sourceRoot":"","sources":["../src/DHTDataBase.ts"],"names":[],"mappings":";;AAAA,OAAO,YAAY,MAAM,QAAQ,CAAC;AAiBlC,MAAM,CAAC,OAAO,OAAO,WAAY,SAAQ,YAAY;IACnD,SAAS,CAAC,KAAK,EAAE,GAAG,CAAC;IACrB,OAAO,CAAC,EAAE,CAAS;IACnB,OAAO,CAAC,YAAY,CAAU;IAC9B,OAAO,CAAC,KAAK,CAAM;IACnB,OAAO,CAAC,WAAW,CAAW;gBAElB,EAAE,EAAE,MAAM,EAAE,EAAE,KAAK,EAAE,GAAE;QAAE,KAAK,CAAC,EAAE,GAAG,CAAA;KAAO;IAavD,OAAO,CAAC,KAAK,CAAK;IAElB,IAAI,IAAI,IAAI,EAAE,CAEb;IAED,IAAI,IAAI,CAAC,KAAK,EAAE,EAAE,EAcjB;IAED,OAAO,CAAC,OAAO,CAAW;IAE1B,IAAI,MAAM,IAAI,MAAM,EAAE,CAErB;IAED,SAAS,CAAC,IAAI,EAAE,GAAG,EAAE,GAAG,CAAC,EAAE,MAAM;IAIjC,UAAU,CAAC,EAAE,EAAE,MAAM,GAAG,MAAM;IAI9B,YAAY,CAAC,EAAE,EAAE,MAAM,GAAG,MAAM,EAAE,KAAK,EAAE,MAAM;IAI/C,WAAW,CAAC,EAAE,EAAE,MAAM,GAAG,MAAM;IAG/B,gBAAgB,CAAC,EAAE,EAAE,MAAM,GAAG,MAAM;IAIpC,gBAAgB,CAAC,EAAE,EAAE,MAAM,GAAG,MAAM;IAIpC,SAAS,CAAC,SAAS,CAAC,EAAE,EAAE,MAAM;IAqC9B,SAAS,CAAC,YAAY,CAAC,EAAE,EAAE,MAAM;IAiBjC,SAAS,CAAC,cAAc,CAAC,OAAO,EAAE,MAAM,EAAE,EAAE,EAAE,MAAM;IAqDpD,OAAO,CAAC,cAAc;IAmBtB,OAAO,CAAC,YAAY;IAIpB,OAAO,CAAC,QAAQ;IAahB,OAAO,CAAC,WAAW;IAOnB,OAAO,CAAC,kBAAkB;IAY1B,OAAO,CAAC,qBAAqB;IAa7B,OAAO,CAAC,cAAc;IAuCtB,OAAO,CAAC,iBAAiB;IAgBzB,OAAO,CAAC,YAAY;IAwBpB,OAAO,CAAC,YAAY;CAOrB"}

14
dist/DHTDataBase.d.ts → dist/dhtOnlineBase.d.ts vendored
View File

@ -1,14 +1,14 @@
/// <reference types="node" />
/// <reference types="node" />
import EventEmitter from "events";
export default class DHTDataBase extends EventEmitter {
protected swarm: any;
export default class DHTOnlineBase extends EventEmitter {
private id;
private bootstrapped;
private graph;
private connectedTo;
constructor(id: Buffer, { swarm }?: {
swarm?: any;
private encoding;
constructor(id: Buffer, { encoding }?: {
encoding?: string | undefined;
});
private _data;
get data(): {};
@ -16,11 +16,7 @@ export default class DHTDataBase extends EventEmitter {
private _online;
get online(): string[];
broadcast(data: any, ttl?: number): void;
getPeerRaw(id: Buffer | string): any;
getPeerField(id: Buffer | string, field: string): any;
getPeerData(id: Buffer | string): any;
getPeerTimestamp(id: Buffer | string): any;
getPeerSignature(id: Buffer | string): any;
protected onAddPeer(id: Buffer): void;
protected onRemovePeer(id: Buffer): void;
protected onGetBroadcast(message: Buffer, id: Buffer): void;
@ -35,4 +31,4 @@ export default class DHTDataBase extends EventEmitter {
private _recalculate;
private _maybeHexify;
}
//# sourceMappingURL=DHTDataBase.d.ts.map
//# sourceMappingURL=dhtOnlineBase.d.ts.map

1
dist/dhtOnlineBase.d.ts.map vendored Normal file
View File

@ -0,0 +1 @@
{"version":3,"file":"dhtOnlineBase.d.ts","sourceRoot":"","sources":["../src/dhtOnlineBase.ts"],"names":[],"mappings":";;AAAA,OAAO,YAAY,MAAM,QAAQ,CAAC;AAYlC,MAAM,CAAC,OAAO,OAAO,aAAc,SAAQ,YAAY;IACrD,OAAO,CAAC,EAAE,CAAS;IACnB,OAAO,CAAC,YAAY,CAAU;IAC9B,OAAO,CAAC,KAAK,CAAM;IACnB,OAAO,CAAC,WAAW,CAAW;IAC9B,OAAO,CAAC,QAAQ,CAAoB;gBAExB,EAAE,EAAE,MAAM,EAAE,EAAE,QAA2B,EAAE;;KAAK;IAa5D,OAAO,CAAC,KAAK,CAAK;IAElB,IAAI,IAAI,IAAI,EAAE,CAEb;IAED,IAAI,IAAI,CAAC,KAAK,EAAE,EAAE,EAMjB;IAED,OAAO,CAAC,OAAO,CAAW;IAE1B,IAAI,MAAM,IAAI,MAAM,EAAE,CAErB;IAED,SAAS,CAAC,IAAI,EAAE,GAAG,EAAE,GAAG,CAAC,EAAE,MAAM;IAIjC,WAAW,CAAC,EAAE,EAAE,MAAM,GAAG,MAAM;IAI/B,SAAS,CAAC,SAAS,CAAC,EAAE,EAAE,MAAM;IAqC9B,SAAS,CAAC,YAAY,CAAC,EAAE,EAAE,MAAM;IAiBjC,SAAS,CAAC,cAAc,CAAC,OAAO,EAAE,MAAM,EAAE,EAAE,EAAE,MAAM;IA6CpD,OAAO,CAAC,cAAc;IAgBtB,OAAO,CAAC,YAAY;IAIpB,OAAO,CAAC,QAAQ;IAGhB,OAAO,CAAC,WAAW;IAOnB,OAAO,CAAC,kBAAkB;IAY1B,OAAO,CAAC,qBAAqB;IAa7B,OAAO,CAAC,cAAc;IA2BtB,OAAO,CAAC,iBAAiB;IAczB,OAAO,CAAC,YAAY;IAwBpB,OAAO,CAAC,YAAY;CAOrB"}

103
dist/DHTDataBase.js → dist/dhtOnlineBase.js vendored
View File

@ -6,20 +6,16 @@ Object.defineProperty(exports, "__esModule", { value: true });
const events_1 = __importDefault(require("events"));
// @ts-ignore
const jsnetworkx_1 = require("jsnetworkx");
// @ts-ignore
const ordered_json_1 = __importDefault(require("ordered-json"));
// @ts-ignore
const hypercore_crypto_1 = __importDefault(require("hypercore-crypto"));
const b4a_1 = __importDefault(require("b4a"));
const codecs_1 = __importDefault(require("codecs"));
const messages_js_1 = require("./messages.js");
const debug_1 = __importDefault(require("debug"));
class DHTDataBase extends events_1.default {
swarm;
const DEFAULT_ENCODING = "json";
class DHTOnlineBase extends events_1.default {
id;
bootstrapped;
graph;
connectedTo;
constructor(id, { swarm } = {}) {
encoding;
constructor(id, { encoding = DEFAULT_ENCODING } = {}) {
super();
if (!id)
throw new TypeError("Must provide id for self");
@ -28,8 +24,8 @@ class DHTDataBase extends events_1.default {
this.graph = new jsnetworkx_1.DiGraph();
this.connectedTo = new Set();
this._data = {};
this.encoding = (0, codecs_1.default)(encoding || DEFAULT_ENCODING);
this._online = [this._maybeHexify(this.id)];
this.swarm = swarm;
}
_data;
get data() {
@ -37,10 +33,7 @@ class DHTDataBase extends events_1.default {
}
set data(value) {
this._data = value;
const timestamp = BigInt(Date.now());
const rawData = ordered_json_1.default.stringify(value);
const signature = hypercore_crypto_1.default.sign(b4a_1.default.from(`${timestamp}${rawData}`), this.swarm.keyPair.secretKey);
this._setPeer(this.id, value, timestamp, signature);
this._setPeer(this.id, value);
this._broadcastData();
}
_online;
@ -50,20 +43,8 @@ class DHTDataBase extends events_1.default {
broadcast(data, ttl) {
throw new TypeError("Broadcast has not been implemented");
}
getPeerRaw(id) {
return this.graph.node.get(this._maybeHexify(id));
}
getPeerField(id, field) {
return this.getPeerRaw(id)?.[field];
}
getPeerData(id) {
return this.getPeerField(id, "data");
}
getPeerTimestamp(id) {
return this.getPeerField(id, "timestamp");
}
getPeerSignature(id) {
return this.getPeerField(id, "signature");
return this.graph.node.get(this._maybeHexify(id));
}
onAddPeer(id) {
const stringId = id.toString("hex");
@ -110,16 +91,11 @@ class DHTDataBase extends events_1.default {
throw new Error("Missing Type In Message");
}
if (type === messages_js_1.Type.STATE) {
const { data: rawData, timestamp, signature } = decoded;
if (signature &&
hypercore_crypto_1.default.verify(b4a_1.default.from(`${timestamp}${rawData}`), signature, id)) {
const data = rawData ? ordered_json_1.default.parse(rawData) : null;
this._setPeer(id, data, timestamp, signature);
this.emit("peer-data", data, id);
this._recalculate();
return;
}
(0, debug_1.default)(`Invalid signature received for peer ${id}`);
const { data: rawData } = decoded;
const data = this.encoding.decode(rawData);
this._setPeer(id, data);
this.emit("peer-data", data, id);
this._recalculate();
}
else if (type === messages_js_1.Type.CONNECTED) {
const { id: toId } = decoded;
@ -150,24 +126,17 @@ class DHTDataBase extends events_1.default {
if (!Object.keys(rawData).length) {
return;
}
const data = ordered_json_1.default.stringify(rawData);
const { timestamp, signature } = this.getPeerRaw(this.id);
const data = this.encoding.encode(rawData);
this.broadcast(messages_js_1.Message.toBinary(messages_js_1.Message.create({
type: messages_js_1.Type.STATE,
data: b4a_1.default.from(data),
signature,
timestamp,
data,
})));
}
_hasSeenPeer(id) {
return this.graph.hasNode(this._maybeHexify(id));
}
_setPeer(id, data, timestamp, signature) {
this.graph.addNode(this._maybeHexify(id), {
timestamp,
signature,
data,
});
_setPeer(id, data) {
this.graph.addNode(this._maybeHexify(id), data);
}
_ensurePeer(id) {
id = this._maybeHexify(id);
@ -196,26 +165,19 @@ class DHTDataBase extends events_1.default {
return;
}
for (const id in bootstrap) {
const { data, connectedTo, signature, timestamp } = bootstrap[id];
if (id === this.id.toString("hex")) {
const { data, connectedTo } = bootstrap[id];
const parsedData = data ? this.encoding.decode(data) : null;
let peerData = parsedData || {};
if (id === this.id.toString("hex"))
continue;
// If we're already tracking them
if (this._hasSeenPeer(id)) {
// See what data we already have for them
// Add their existing data to what we got from the bootstrap
const existingPeerData = this.getPeerData(id);
peerData = { ...existingPeerData, ...peerData };
}
if (signature &&
hypercore_crypto_1.default.verify(b4a_1.default.from(`${timestamp}${data}`), signature, b4a_1.default.from(id))) {
const parsedData = data ? ordered_json_1.default.parse(data) : null;
let peerData = parsedData || {};
// If we're already tracking them
if (this._hasSeenPeer(id)) {
// See what data we already have for them
// Add their existing data to what we got from the bootstrap
const existingPeerData = this.getPeerData(id);
peerData = { ...existingPeerData, ...peerData };
this._setPeer(id, peerData, timestamp, signature);
}
}
else {
(0, debug_1.default)(`Invalid signature received for peer ${id}`);
}
this._setPeer(id, peerData);
for (const connection of connectedTo) {
this._addPeerConnection(id, Buffer.from(connection));
}
@ -229,9 +191,8 @@ class DHTDataBase extends events_1.default {
const connectedTo = this.graph
.neighbors(id)
.map((id) => Buffer.from(id, "hex"));
const data = rawData ? ordered_json_1.default.stringify(rawData?.data) : null;
const { timestamp = undefined, signature = undefined } = rawData;
state[id] = { data: b4a_1.default.from(data), connectedTo, timestamp, signature };
const data = rawData ? this.encoding.encode(rawData) : null;
state[id] = { data, connectedTo };
}
return state;
}
@ -256,10 +217,10 @@ class DHTDataBase extends events_1.default {
this.emit("online", online);
}
_maybeHexify(data) {
if (b4a_1.default.isBuffer(data)) {
if (Buffer.isBuffer(data)) {
return data.toString("hex");
}
return data;
}
}
exports.default = DHTDataBase;
exports.default = DHTOnlineBase;

5
dist/index.d.ts vendored
View File

@ -1,7 +1,8 @@
/// <reference types="node" />
import DHTDataBase from "./DHTDataBase.js";
export default class DHTData extends DHTDataBase {
import DHTOnlineBase from "./dhtOnlineBase.js";
export default class DHTOnline extends DHTOnlineBase {
private flood;
private swarm;
constructor(swarm: any, { id, data, ...opts }?: {
id?: Buffer;
data?: {};

2
dist/index.d.ts.map vendored
View File

@ -1 +1 @@
{"version":3,"file":"index.d.ts","sourceRoot":"","sources":["../src/index.ts"],"names":[],"mappings":";AAAA,OAAO,WAAW,MAAM,kBAAkB,CAAC;AAK3C,MAAM,CAAC,OAAO,OAAO,OAAQ,SAAQ,WAAW;IAC9C,OAAO,CAAC,KAAK,CAAW;gBAGtB,KAAK,EAAE,GAAG,EACV,EACE,EAA4B,EAC5B,IAAS,EACT,GAAG,IAAI,EACR,GAAE;QAAE,EAAE,CAAC,EAAE,MAAM,CAAC;QAAC,IAAI,CAAC,EAAE,EAAE,CAAC;QAAC,CAAC,GAAG,EAAE,MAAM,GAAG,GAAG,CAAA;KAAO;IAkBxD,aAAa,CAAC,IAAI,EAAE,GAAG;IAKvB,gBAAgB,CAAC,IAAI,EAAE,GAAG;IAW1B,SAAS,CAAC,OAAO,EAAE,GAAG,EAAE,GAAG,CAAC,EAAE,MAAM;CAGrC"}
{"version":3,"file":"index.d.ts","sourceRoot":"","sources":["../src/index.ts"],"names":[],"mappings":";AAAA,OAAO,aAAa,MAAM,oBAAoB,CAAC;AAK/C,MAAM,CAAC,OAAO,OAAO,SAAU,SAAQ,aAAa;IAClD,OAAO,CAAC,KAAK,CAAW;IACxB,OAAO,CAAC,KAAK,CAAM;gBAGjB,KAAK,EAAE,GAAG,EACV,EACE,EAA4B,EAC5B,IAAS,EACT,GAAG,IAAI,EACR,GAAE;QAAE,EAAE,CAAC,EAAE,MAAM,CAAC;QAAC,IAAI,CAAC,EAAE,EAAE,CAAC;QAAC,CAAC,GAAG,EAAE,MAAM,GAAG,GAAG,CAAA;KAAO;IAmBxD,aAAa,CAAC,IAAI,EAAE,GAAG;IAKvB,gBAAgB,CAAC,IAAI,EAAE,GAAG;IAW1B,SAAS,CAAC,OAAO,EAAE,GAAG,EAAE,GAAG,CAAC,EAAE,MAAM;CAGrC"}

10
dist/index.js vendored
View File

@ -3,14 +3,16 @@ var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
const DHTDataBase_js_1 = __importDefault(require("./DHTDataBase.js"));
const dhtOnlineBase_js_1 = __importDefault(require("./dhtOnlineBase.js"));
const dht_flood_1 = __importDefault(require("@lumeweb/dht-flood"));
const DISCONNECT_SMOOTH = 500;
class DHTData extends DHTDataBase_js_1.default {
class DHTOnline extends dhtOnlineBase_js_1.default {
flood;
swarm;
constructor(swarm, { id = swarm.keyPair.publicKey, data = {}, ...opts } = {}) {
super(id, { swarm, ...opts });
super(id, opts);
this.flood = new dht_flood_1.default({ id, swarm, ...opts });
this.swarm = swarm;
this.flood.on("peer-open", (peer) => this.handlePeerAdd(peer));
this.flood.on("peer-remove", (peer) => this.handlePeerRemove(peer));
this.flood.on("message", (message, id) => this.onGetBroadcast(message, id));
@ -36,4 +38,4 @@ class DHTData extends DHTDataBase_js_1.default {
this.flood.broadcast(message, ttl);
}
}
exports.default = DHTData;
exports.default = DHTOnline;

18
dist/messages.d.ts vendored
View File

@ -23,15 +23,7 @@ export interface Message {
*/
data?: Uint8Array;
/**
* @generated from protobuf field: optional bytes signature = 4;
*/
signature?: Uint8Array;
/**
* @generated from protobuf field: optional int64 timestamp = 5;
*/
timestamp?: bigint;
/**
* @generated from protobuf field: optional bytes id = 6;
* @generated from protobuf field: optional bytes id = 4;
*/
id?: Uint8Array;
}
@ -47,14 +39,6 @@ export interface State {
* @generated from protobuf field: optional bytes data = 2;
*/
data?: Uint8Array;
/**
* @generated from protobuf field: optional bytes signature = 3;
*/
signature?: Uint8Array;
/**
* @generated from protobuf field: optional int64 timestamp = 4;
*/
timestamp?: bigint;
}
/**
* @generated from protobuf enum Type

2
dist/messages.d.ts.map vendored
View File

@ -1 +1 @@
{"version":3,"file":"messages.d.ts","sourceRoot":"","sources":["../src/messages.ts"],"names":[],"mappings":"AAGA,OAAO,KAAK,EAAE,kBAAkB,EAAE,MAAM,sBAAsB,CAAC;AAC/D,OAAO,KAAK,EAAE,aAAa,EAAE,MAAM,sBAAsB,CAAC;AAE1D,OAAO,KAAK,EAAE,iBAAiB,EAAE,MAAM,sBAAsB,CAAC;AAC9D,OAAO,KAAK,EAAE,aAAa,EAAE,MAAM,sBAAsB,CAAC;AAE1D,OAAO,KAAK,EAAE,cAAc,EAAE,MAAM,sBAAsB,CAAC;AAG3D,OAAO,EAAE,WAAW,EAAE,MAAM,sBAAsB,CAAC;AACnD;;GAEG;AACH,MAAM,WAAW,OAAO;IACpB;;OAEG;IACH,IAAI,EAAE,IAAI,CAAC;IACX;;OAEG;IACH,SAAS,EAAE;QACP,CAAC,GAAG,EAAE,MAAM,GAAG,KAAK,CAAC;KACxB,CAAC;IACF;;OAEG;IACH,IAAI,CAAC,EAAE,UAAU,CAAC;IAClB;;OAEG;IACH,SAAS,CAAC,EAAE,UAAU,CAAC;IACvB;;OAEG;IACH,SAAS,CAAC,EAAE,MAAM,CAAC;IACnB;;OAEG;IACH,EAAE,CAAC,EAAE,UAAU,CAAC;CACnB;AACD;;GAEG;AACH,MAAM,WAAW,KAAK;IAClB;;OAEG;IACH,WAAW,EAAE,UAAU,EAAE,CAAC;IAC1B;;OAEG;IACH,IAAI,CAAC,EAAE,UAAU,CAAC;IAClB;;OAEG;IACH,SAAS,CAAC,EAAE,UAAU,CAAC;IACvB;;OAEG;IACH,SAAS,CAAC,EAAE,MAAM,CAAC;CACtB;AACD;;GAEG;AACH,oBAAY,IAAI;IACZ;;OAEG;IACH,YAAY,IAAI;IAChB;;OAEG;IACH,iBAAiB,IAAI;IACrB;;OAEG;IACH,kBAAkB,IAAI;IACtB;;OAEG;IACH,SAAS,IAAI;IACb;;OAEG;IACH,YAAY,IAAI;IAChB;;OAEG;IACH,KAAK,IAAI;CACZ;AAED,cAAM,YAAa,SAAQ,WAAW,CAAC,OAAO,CAAC;;IAW3C,MAAM,CAAC,KAAK,CAAC,EAAE,cAAc,CAAC,OAAO,CAAC,GAAG,OAAO;IAOhD,kBAAkB,CAAC,MAAM,EAAE,aAAa,EAAE,MAAM,EAAE,MAAM,EAAE,OAAO,EAAE,iBAAiB,EAAE,MAAM,CAAC,EAAE,OAAO,GAAG,OAAO;IAkChH,OAAO,CAAC,cAAc;IAgBtB,mBAAmB,CAAC,OAAO,EAAE,OAAO,EAAE,MAAM,EAAE,aAAa,EAAE,OAAO,EAAE,kBAAkB,GAAG,aAAa;CA4B3G;AACD;;GAEG;AACH,eAAO,MAAM,OAAO,cAAqB,CAAC;AAE1C,cAAM,UAAW,SAAQ,WAAW,CAAC,KAAK,CAAC;;IASvC,MAAM,CAAC,KAAK,CAAC,EAAE,cAAc,CAAC,KAAK,CAAC,GAAG,KAAK;IAO5C,kBAAkB,CAAC,MAAM,EAAE,aAAa,EAAE,MAAM,EAAE,MAAM,EAAE,OAAO,EAAE,iBAAiB,EAAE,MAAM,CAAC,EAAE,KAAK,GAAG,KAAK;IA4B5G,mBAAmB,CAAC,OAAO,EAAE,KAAK,EAAE,MAAM,EAAE,aAAa,EAAE,OAAO,EAAE,kBAAkB,GAAG,aAAa;CAkBzG;AACD;;GAEG;AACH,eAAO,MAAM,KAAK,YAAmB,CAAC"}
{"version":3,"file":"messages.d.ts","sourceRoot":"","sources":["../src/messages.ts"],"names":[],"mappings":"AAGA,OAAO,KAAK,EAAE,kBAAkB,EAAE,MAAM,sBAAsB,CAAC;AAC/D,OAAO,KAAK,EAAE,aAAa,EAAE,MAAM,sBAAsB,CAAC;AAE1D,OAAO,KAAK,EAAE,iBAAiB,EAAE,MAAM,sBAAsB,CAAC;AAC9D,OAAO,KAAK,EAAE,aAAa,EAAE,MAAM,sBAAsB,CAAC;AAE1D,OAAO,KAAK,EAAE,cAAc,EAAE,MAAM,sBAAsB,CAAC;AAG3D,OAAO,EAAE,WAAW,EAAE,MAAM,sBAAsB,CAAC;AACnD;;GAEG;AACH,MAAM,WAAW,OAAO;IACpB;;OAEG;IACH,IAAI,EAAE,IAAI,CAAC;IACX;;OAEG;IACH,SAAS,EAAE;QACP,CAAC,GAAG,EAAE,MAAM,GAAG,KAAK,CAAC;KACxB,CAAC;IACF;;OAEG;IACH,IAAI,CAAC,EAAE,UAAU,CAAC;IAClB;;OAEG;IACH,EAAE,CAAC,EAAE,UAAU,CAAC;CACnB;AACD;;GAEG;AACH,MAAM,WAAW,KAAK;IAClB;;OAEG;IACH,WAAW,EAAE,UAAU,EAAE,CAAC;IAC1B;;OAEG;IACH,IAAI,CAAC,EAAE,UAAU,CAAC;CACrB;AACD;;GAEG;AACH,oBAAY,IAAI;IACZ;;OAEG;IACH,YAAY,IAAI;IAChB;;OAEG;IACH,iBAAiB,IAAI;IACrB;;OAEG;IACH,kBAAkB,IAAI;IACtB;;OAEG;IACH,SAAS,IAAI;IACb;;OAEG;IACH,YAAY,IAAI;IAChB;;OAEG;IACH,KAAK,IAAI;CACZ;AAED,cAAM,YAAa,SAAQ,WAAW,CAAC,OAAO,CAAC;;IAS3C,MAAM,CAAC,KAAK,CAAC,EAAE,cAAc,CAAC,OAAO,CAAC,GAAG,OAAO;IAOhD,kBAAkB,CAAC,MAAM,EAAE,aAAa,EAAE,MAAM,EAAE,MAAM,EAAE,OAAO,EAAE,iBAAiB,EAAE,MAAM,CAAC,EAAE,OAAO,GAAG,OAAO;IA4BhH,OAAO,CAAC,cAAc;IAgBtB,mBAAmB,CAAC,OAAO,EAAE,OAAO,EAAE,MAAM,EAAE,aAAa,EAAE,OAAO,EAAE,kBAAkB,GAAG,aAAa;CAsB3G;AACD;;GAEG;AACH,eAAO,MAAM,OAAO,cAAqB,CAAC;AAE1C,cAAM,UAAW,SAAQ,WAAW,CAAC,KAAK,CAAC;;IAOvC,MAAM,CAAC,KAAK,CAAC,EAAE,cAAc,CAAC,KAAK,CAAC,GAAG,KAAK;IAO5C,kBAAkB,CAAC,MAAM,EAAE,aAAa,EAAE,MAAM,EAAE,MAAM,EAAE,OAAO,EAAE,iBAAiB,EAAE,MAAM,CAAC,EAAE,KAAK,GAAG,KAAK;IAsB5G,mBAAmB,CAAC,OAAO,EAAE,KAAK,EAAE,MAAM,EAAE,aAAa,EAAE,OAAO,EAAE,kBAAkB,GAAG,aAAa;CAYzG;AACD;;GAEG;AACH,eAAO,MAAM,KAAK,YAAmB,CAAC"}

38
dist/messages.js vendored
View File

@ -43,9 +43,7 @@ class Message$Type extends runtime_5.MessageType {
{ no: 1, name: "type", kind: "enum", T: () => ["Type", Type] },
{ no: 2, name: "bootstrap", kind: "map", K: 9 /*ScalarType.STRING*/, V: { kind: "message", T: () => exports.State } },
{ no: 3, name: "data", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ },
{ no: 4, name: "signature", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ },
{ no: 5, name: "timestamp", kind: "scalar", opt: true, T: 3 /*ScalarType.INT64*/, L: 0 /*LongType.BIGINT*/ },
{ no: 6, name: "id", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ }
{ no: 4, name: "id", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ }
]);
}
create(value) {
@ -69,13 +67,7 @@ class Message$Type extends runtime_5.MessageType {
case /* optional bytes data */ 3:
message.data = reader.bytes();
break;
case /* optional bytes signature */ 4:
message.signature = reader.bytes();
break;
case /* optional int64 timestamp */ 5:
message.timestamp = reader.int64().toBigInt();
break;
case /* optional bytes id */ 6:
case /* optional bytes id */ 4:
message.id = reader.bytes();
break;
default:
@ -119,15 +111,9 @@ class Message$Type extends runtime_5.MessageType {
/* optional bytes data = 3; */
if (message.data !== undefined)
writer.tag(3, runtime_1.WireType.LengthDelimited).bytes(message.data);
/* optional bytes signature = 4; */
if (message.signature !== undefined)
writer.tag(4, runtime_1.WireType.LengthDelimited).bytes(message.signature);
/* optional int64 timestamp = 5; */
if (message.timestamp !== undefined)
writer.tag(5, runtime_1.WireType.Varint).int64(message.timestamp);
/* optional bytes id = 6; */
/* optional bytes id = 4; */
if (message.id !== undefined)
writer.tag(6, runtime_1.WireType.LengthDelimited).bytes(message.id);
writer.tag(4, runtime_1.WireType.LengthDelimited).bytes(message.id);
let u = options.writeUnknownFields;
if (u !== false)
(u == true ? runtime_2.UnknownFieldHandler.onWrite : u)(this.typeName, message, writer);
@ -143,9 +129,7 @@ class State$Type extends runtime_5.MessageType {
constructor() {
super("State", [
{ no: 1, name: "connectedTo", kind: "scalar", repeat: 2 /*RepeatType.UNPACKED*/, T: 12 /*ScalarType.BYTES*/ },
{ no: 2, name: "data", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ },
{ no: 3, name: "signature", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ },
{ no: 4, name: "timestamp", kind: "scalar", opt: true, T: 3 /*ScalarType.INT64*/, L: 0 /*LongType.BIGINT*/ }
{ no: 2, name: "data", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ }
]);
}
create(value) {
@ -166,12 +150,6 @@ class State$Type extends runtime_5.MessageType {
case /* optional bytes data */ 2:
message.data = reader.bytes();
break;
case /* optional bytes signature */ 3:
message.signature = reader.bytes();
break;
case /* optional int64 timestamp */ 4:
message.timestamp = reader.int64().toBigInt();
break;
default:
let u = options.readUnknownField;
if (u === "throw")
@ -190,12 +168,6 @@ class State$Type extends runtime_5.MessageType {
/* optional bytes data = 2; */
if (message.data !== undefined)
writer.tag(2, runtime_1.WireType.LengthDelimited).bytes(message.data);
/* optional bytes signature = 3; */
if (message.signature !== undefined)
writer.tag(3, runtime_1.WireType.LengthDelimited).bytes(message.signature);
/* optional int64 timestamp = 4; */
if (message.timestamp !== undefined)
writer.tag(4, runtime_1.WireType.Varint).int64(message.timestamp);
let u = options.writeUnknownFields;
if (u !== false)
(u == true ? runtime_2.UnknownFieldHandler.onWrite : u)(this.typeName, message, writer);

26
messages.proto
View File

@ -1,25 +1,21 @@
syntax = "proto2";
enum Type {
BOOTSTRAP_REQUEST = 1;
BOOTSTRAP_RESPONSE = 2;
CONNECTED = 3;
DISCONNECTED = 4;
STATE = 5;
BOOTSTRAP_REQUEST = 1;
BOOTSTRAP_RESPONSE = 2;
CONNECTED = 3;
DISCONNECTED = 4;
STATE = 5;
}
message Message {
required Type type = 1;
map<string, State> bootstrap = 2; // For bootstrap events
optional bytes data = 3; // For state event
optional bytes signature = 4;
optional int64 timestamp = 5;
optional bytes id = 6; // For connected and disconnected events
required Type type = 1;
map<string, State> bootstrap = 2; // For bootstrap events
optional bytes data = 3; // For state event
optional bytes id = 4; // For connected and disconnected events
}
message State {
repeated bytes connectedTo = 1;
optional bytes data = 2;
optional bytes signature = 3;
optional int64 timestamp = 4;
repeated bytes connectedTo = 1;
optional bytes data = 2;
}

7
package.json
View File

@ -1,17 +1,16 @@
{
"name": "@lumeweb/dht-data",
"name": "@lumeweb/dht-online",
"type": "commonjs",
"version": "0.1.0",
"main": "dist/index.js",
"dependencies": {
"@lumeweb/dht-flood": "https://git.lumeweb.com/LumeWeb/dht-flood.git",
"@protobuf-ts/plugin": "^2.8.1",
"b4a": "^1.6.1",
"@types/codecs": "^2.2.3",
"codecs": "^3.0.0",
"compact-encoding": "^2.11.0",
"hypercore-crypto": "^3.3.0",
"jsnetworkx": "^0.3.4",
"lru": "^3.1.0",
"ordered-json": "^0.1.1",
"protocol-buffers-encodings": "^1.2.0",
"protomux-rpc": "^1.3.0"
},

127
src/DHTDataBase.ts → src/dhtOnlineBase.ts
View File

@ -1,28 +1,23 @@
import EventEmitter from "events";
// @ts-ignore
import { DiGraph, hasPath } from "jsnetworkx";
// @ts-ignore
import orderedJSON from "ordered-json";
// @ts-ignore
import crypto from "hypercore-crypto";
import b4a from "b4a";
import codecs from "codecs";
import { Message, State, Type } from "./messages.js";
// @ts-ignore
import sodium from "sodium-universal";
import debug from "debug";
const DEFAULT_ENCODING = "json";
type Bootstrap = {
[key: string]: State;
};
export default class DHTDataBase extends EventEmitter {
protected swarm: any;
export default class DHTOnlineBase extends EventEmitter {
private id: Buffer;
private bootstrapped: boolean;
private graph: any;
private connectedTo: Set<any>;
private encoding: codecs.Codec<any>;
constructor(id: Buffer, { swarm }: { swarm?: any } = {}) {
constructor(id: Buffer, { encoding = DEFAULT_ENCODING } = {}) {
super();
if (!id) throw new TypeError("Must provide id for self");
@ -31,8 +26,8 @@ export default class DHTDataBase extends EventEmitter {
this.graph = new DiGraph();
this.connectedTo = new Set();
this._data = {};
this.encoding = codecs(encoding || DEFAULT_ENCODING);
this._online = [this._maybeHexify(this.id)];
this.swarm = swarm;
}
private _data: {};
@ -44,16 +39,8 @@ export default class DHTDataBase extends EventEmitter {
set data(value: {}) {
this._data = value;
const timestamp = BigInt(Date.now());
this._setPeer(this.id, value);
const rawData = orderedJSON.stringify(value);
const signature = crypto.sign(
b4a.from(`${timestamp}${rawData}`),
this.swarm.keyPair.secretKey
);
this._setPeer(this.id, value, timestamp, signature);
this._broadcastData();
}
@ -67,23 +54,8 @@ export default class DHTDataBase extends EventEmitter {
throw new TypeError("Broadcast has not been implemented");
}
getPeerRaw(id: Buffer | string) {
return this.graph.node.get(this._maybeHexify(id));
}
getPeerField(id: Buffer | string, field: string) {
return this.getPeerRaw(id)?.[field];
}
getPeerData(id: Buffer | string) {
return this.getPeerField(id, "data");
}
getPeerTimestamp(id: Buffer | string) {
return this.getPeerField(id, "timestamp");
}
getPeerSignature(id: Buffer | string) {
return this.getPeerField(id, "signature");
return this.graph.node.get(this._maybeHexify(id));
}
protected onAddPeer(id: Buffer) {
@ -153,19 +125,11 @@ export default class DHTDataBase extends EventEmitter {
throw new Error("Missing Type In Message");
}
if (type === Type.STATE) {
const { data: rawData, timestamp, signature } = decoded;
if (
signature &&
crypto.verify(b4a.from(`${timestamp}${rawData}`), signature, id)
) {
const data = rawData ? orderedJSON.parse(rawData) : null;
this._setPeer(id, data, timestamp, signature);
this.emit("peer-data", data, id);
this._recalculate();
return;
}
debug(`Invalid signature received for peer ${id}`);
const { data: rawData } = decoded;
const data = this.encoding.decode(rawData);
this._setPeer(id, data);
this.emit("peer-data", data, id);
this._recalculate();
} else if (type === Type.CONNECTED) {
const { id: toId } = decoded;
this._addPeerConnection(id, Buffer.from(toId as Uint8Array));
@ -198,15 +162,12 @@ export default class DHTDataBase extends EventEmitter {
if (!Object.keys(rawData).length) {
return;
}
const data = orderedJSON.stringify(rawData);
const { timestamp, signature } = this.getPeerRaw(this.id);
const data = this.encoding.encode(rawData);
this.broadcast(
Message.toBinary(
Message.create({
type: Type.STATE,
data: b4a.from(data),
signature,
timestamp,
data,
})
)
);
@ -216,19 +177,9 @@ export default class DHTDataBase extends EventEmitter {
return this.graph.hasNode(this._maybeHexify(id));
}
private _setPeer(
id: Buffer | string,
data: any,
timestamp?: BigInt,
signature?: Uint8Array
) {
this.graph.addNode(this._maybeHexify(id), {
timestamp,
signature,
data,
});
private _setPeer(id: Buffer | string, data: any) {
this.graph.addNode(this._maybeHexify(id), data);
}
private _ensurePeer(id: Buffer | string) {
id = this._maybeHexify(id);
if (!this._hasSeenPeer(id)) {
@ -267,29 +218,18 @@ export default class DHTDataBase extends EventEmitter {
}
for (const id in bootstrap) {
const { data, connectedTo, signature, timestamp } = bootstrap[id];
if (id === this.id.toString("hex")) {
continue;
}
if (
signature &&
crypto.verify(b4a.from(`${timestamp}${data}`), signature, b4a.from(id))
) {
const parsedData = data ? orderedJSON.parse(data) : null;
let peerData = parsedData || {};
// If we're already tracking them
if (this._hasSeenPeer(id)) {
// See what data we already have for them
// Add their existing data to what we got from the bootstrap
const existingPeerData = this.getPeerData(id);
peerData = { ...existingPeerData, ...peerData };
this._setPeer(id, peerData, timestamp, signature);
}
} else {
debug(`Invalid signature received for peer ${id}`);
const { data, connectedTo } = bootstrap[id];
const parsedData = data ? this.encoding.decode(data) : null;
let peerData = parsedData || {};
if (id === this.id.toString("hex")) continue;
// If we're already tracking them
if (this._hasSeenPeer(id)) {
// See what data we already have for them
// Add their existing data to what we got from the bootstrap
const existingPeerData = this.getPeerData(id);
peerData = { ...existingPeerData, ...peerData };
}
this._setPeer(id, peerData);
for (const connection of connectedTo) {
this._addPeerConnection(id, Buffer.from(connection));
}
@ -299,17 +239,14 @@ export default class DHTDataBase extends EventEmitter {
this._recalculate();
}
private _getBootstrapInfo() {
const state: Bootstrap = {};
for (const [id, rawData] of this.graph.nodes(true)) {
const connectedTo = this.graph
.neighbors(id)
.map((id: string) => Buffer.from(id, "hex"));
const data = rawData ? orderedJSON.stringify(rawData?.data) : null;
const { timestamp = undefined, signature = undefined } = rawData;
state[id] = { data: b4a.from(data), connectedTo, timestamp, signature };
const data = rawData ? this.encoding.encode(rawData) : null;
state[id] = { data, connectedTo };
}
return state;
@ -341,7 +278,7 @@ export default class DHTDataBase extends EventEmitter {
}
private _maybeHexify(data: Buffer | string): string {
if (b4a.isBuffer(data)) {
if (Buffer.isBuffer(data)) {
return data.toString("hex");
}

8
src/index.ts
View File

@ -1,10 +1,11 @@
import DHTDataBase from "./DHTDataBase.js";
import DHTOnlineBase from "./dhtOnlineBase.js";
import DHTFlood from "@lumeweb/dht-flood";
const DISCONNECT_SMOOTH = 500;
export default class DHTData extends DHTDataBase {
export default class DHTOnline extends DHTOnlineBase {
private flood: DHTFlood;
private swarm: any;
constructor(
swarm: any,
@ -14,8 +15,9 @@ export default class DHTData extends DHTDataBase {
...opts
}: { id?: Buffer; data?: {}; [key: string]: any } = {}
) {
super(id, { swarm, ...(opts as any) });
super(id, opts as any);
this.flood = new DHTFlood({ id, swarm, ...opts });
this.swarm = swarm;
this.flood.on("peer-open", (peer) => this.handlePeerAdd(peer));
this.flood.on("peer-remove", (peer) => this.handlePeerRemove(peer));

56
src/messages.ts
View File

@ -30,15 +30,7 @@ export interface Message {
*/
data?: Uint8Array; // For state event
/**
* @generated from protobuf field: optional bytes signature = 4;
*/
signature?: Uint8Array;
/**
* @generated from protobuf field: optional int64 timestamp = 5;
*/
timestamp?: bigint;
/**
* @generated from protobuf field: optional bytes id = 6;
* @generated from protobuf field: optional bytes id = 4;
*/
id?: Uint8Array; // For connected and disconnected events
}
@ -54,14 +46,6 @@ export interface State {
* @generated from protobuf field: optional bytes data = 2;
*/
data?: Uint8Array;
/**
* @generated from protobuf field: optional bytes signature = 3;
*/
signature?: Uint8Array;
/**
* @generated from protobuf field: optional int64 timestamp = 4;
*/
timestamp?: bigint;
}
/**
* @generated from protobuf enum Type
@ -99,9 +83,7 @@ class Message$Type extends MessageType<Message> {
{ no: 1, name: "type", kind: "enum", T: () => ["Type", Type] },
{ no: 2, name: "bootstrap", kind: "map", K: 9 /*ScalarType.STRING*/, V: { kind: "message", T: () => State } },
{ no: 3, name: "data", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ },
{ no: 4, name: "signature", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ },
{ no: 5, name: "timestamp", kind: "scalar", opt: true, T: 3 /*ScalarType.INT64*/, L: 0 /*LongType.BIGINT*/ },
{ no: 6, name: "id", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ }
{ no: 4, name: "id", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ }
]);
}
create(value?: PartialMessage<Message>): Message {
@ -125,13 +107,7 @@ class Message$Type extends MessageType<Message> {
case /* optional bytes data */ 3:
message.data = reader.bytes();
break;
case /* optional bytes signature */ 4:
message.signature = reader.bytes();
break;
case /* optional int64 timestamp */ 5:
message.timestamp = reader.int64().toBigInt();
break;
case /* optional bytes id */ 6:
case /* optional bytes id */ 4:
message.id = reader.bytes();
break;
default:
@ -175,15 +151,9 @@ class Message$Type extends MessageType<Message> {
/* optional bytes data = 3; */
if (message.data !== undefined)
writer.tag(3, WireType.LengthDelimited).bytes(message.data);
/* optional bytes signature = 4; */
if (message.signature !== undefined)
writer.tag(4, WireType.LengthDelimited).bytes(message.signature);
/* optional int64 timestamp = 5; */
if (message.timestamp !== undefined)
writer.tag(5, WireType.Varint).int64(message.timestamp);
/* optional bytes id = 6; */
/* optional bytes id = 4; */
if (message.id !== undefined)
writer.tag(6, WireType.LengthDelimited).bytes(message.id);
writer.tag(4, WireType.LengthDelimited).bytes(message.id);
let u = options.writeUnknownFields;
if (u !== false)
(u == true ? UnknownFieldHandler.onWrite : u)(this.typeName, message, writer);
@ -199,9 +169,7 @@ class State$Type extends MessageType<State> {
constructor() {
super("State", [
{ no: 1, name: "connectedTo", kind: "scalar", repeat: 2 /*RepeatType.UNPACKED*/, T: 12 /*ScalarType.BYTES*/ },
{ no: 2, name: "data", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ },
{ no: 3, name: "signature", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ },
{ no: 4, name: "timestamp", kind: "scalar", opt: true, T: 3 /*ScalarType.INT64*/, L: 0 /*LongType.BIGINT*/ }
{ no: 2, name: "data", kind: "scalar", opt: true, T: 12 /*ScalarType.BYTES*/ }
]);
}
create(value?: PartialMessage<State>): State {
@ -222,12 +190,6 @@ class State$Type extends MessageType<State> {
case /* optional bytes data */ 2:
message.data = reader.bytes();
break;
case /* optional bytes signature */ 3:
message.signature = reader.bytes();
break;
case /* optional int64 timestamp */ 4:
message.timestamp = reader.int64().toBigInt();
break;
default:
let u = options.readUnknownField;
if (u === "throw")
@ -246,12 +208,6 @@ class State$Type extends MessageType<State> {
/* optional bytes data = 2; */
if (message.data !== undefined)
writer.tag(2, WireType.LengthDelimited).bytes(message.data);
/* optional bytes signature = 3; */
if (message.signature !== undefined)
writer.tag(3, WireType.LengthDelimited).bytes(message.signature);
/* optional int64 timestamp = 4; */
if (message.timestamp !== undefined)
writer.tag(4, WireType.Varint).int64(message.timestamp);
let u = options.writeUnknownFields;
if (u !== false)
(u == true ? UnknownFieldHandler.onWrite : u)(this.typeName, message, writer);