dht-data/dist/dhtOnlineBase.js

"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
    return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
const events_1 = __importDefault(require("events"));
// @ts-ignore
const jsnetworkx_1 = require("jsnetworkx");
const codecs_1 = __importDefault(require("codecs"));
const messages_js_1 = require("./messages.js");
const DEFAULT_ENCODING = "json";
class DHTOnlineBase extends events_1.default {
    id;
    bootstrapped;
    graph;
    connectedTo;
    encoding;
    constructor(id, { encoding = DEFAULT_ENCODING } = {}) {
        super();
        if (!id)
            throw new TypeError("Must provide id for self");
        this.id = id;
        this.bootstrapped = false;
        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)];
    }
    _data;
    get data() {
        return this._data;
    }
    set data(value) {
        this._data = value;
        this._setPeer(this.id, value);
        this._broadcastData();
    }
    _online;
    get online() {
        return this._online;
    }
    broadcast(data, ttl) {
        throw new TypeError("Broadcast has not been implemented");
    }
    getPeerData(id) {
        return this.graph.node.get(this._maybeHexify(id));
    }
    onAddPeer(id) {
        const stringId = id.toString("hex");
        if (this.connectedTo.has(stringId)) {
            return;
        } // Already know we're connected here
        this.connectedTo.add(stringId);
        this._addPeerConnection(this.id, id);
        this.emit("peer-add", id);
        this._recalculate();
        this.broadcast(messages_js_1.Message.toBinary(messages_js_1.Message.create({
            type: messages_js_1.Type.CONNECTED,
            id,
        })));
        this._broadcastData();
        if (this.bootstrapped) {
            return;
        }
        // If this is the first person we've met, get their graph
        this.broadcast(messages_js_1.Message.toBinary(messages_js_1.Message.create({
            type: messages_js_1.Type.BOOTSTRAP_REQUEST,
        })), 0);
    }
    onRemovePeer(id) {
        this.connectedTo.delete(id.toString("hex"));
        this._removePeerConnection(this.id, id);
        this.emit("peer-remove");
        this._recalculate();
        this.broadcast(messages_js_1.Message.toBinary(messages_js_1.Message.create({
            type: messages_js_1.Type.DISCONNECTED,
            id,
        })));
    }
    onGetBroadcast(message, id) {
        let decoded;
        try {
            decoded = messages_js_1.Message.fromBinary(message);
        }
        catch {
            return;
        }
        const { type } = decoded;
        if (!type) {
            throw new Error("Missing Type In Message");
        }
        if (type === messages_js_1.Type.STATE) {
            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;
            this._addPeerConnection(id, Buffer.from(toId));
            this.emit("peer-add-seen", id, toId);
            this._recalculate();
        }
        else if (type === messages_js_1.Type.DISCONNECTED) {
            const { id: toId } = decoded;
            this._removePeerConnection(id, Buffer.from(toId));
            this.emit("peer-remove-seen", id, toId);
            this._recalculate();
        }
        else if (type === messages_js_1.Type.BOOTSTRAP_REQUEST) {
            const bootstrap = this._getBootstrapInfo();
            this.broadcast(messages_js_1.Message.toBinary(messages_js_1.Message.create({
                type: messages_js_1.Type.BOOTSTRAP_RESPONSE,
                bootstrap,
            })), 0);
        }
        else if (type === messages_js_1.Type.BOOTSTRAP_RESPONSE) {
            const { bootstrap } = decoded;
            this._bootstrapFrom(bootstrap);
        }
    }
    _broadcastData() {
        const rawData = this._data;
        if (!Object.keys(rawData).length) {
            return;
        }
        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,
        })));
    }
    _hasSeenPeer(id) {
        return this.graph.hasNode(this._maybeHexify(id));
    }
    _setPeer(id, data) {
        this.graph.addNode(this._maybeHexify(id), data);
    }
    _ensurePeer(id) {
        id = this._maybeHexify(id);
        if (!this._hasSeenPeer(id)) {
            this._setPeer(id, {});
        }
    }
    _addPeerConnection(origin, destination) {
        this._ensurePeer(origin);
        this._ensurePeer(destination);
        this.graph.addEdge(this._maybeHexify(origin), this._maybeHexify(destination));
    }
    _removePeerConnection(origin, destination) {
        try {
            this._ensurePeer(origin);
            this._ensurePeer(destination);
            this.graph.removeEdge(origin.toString("hex"), destination.toString("hex"));
        }
        catch (e) {
            if (e.name !== "JSNetworkXError")
                throw e;
        }
    }
    _bootstrapFrom(bootstrap) {
        if (this.bootstrapped) {
            return;
        }
        for (const id in bootstrap) {
            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));
            }
        }
        this.emit("bootstrapped");
        this._recalculate();
    }
    _getBootstrapInfo() {
        const state = {};
        for (const [id, rawData] of this.graph.nodes(true)) {
            const connectedTo = this.graph
                .neighbors(id)
                .map((id) => Buffer.from(id, "hex"));
            const data = rawData ? this.encoding.encode(rawData) : null;
            state[id] = { data, connectedTo };
        }
        return state;
    }
    // Calculate who's online and emit an event
    _recalculate() {
        const online = this.graph.nodes().filter((id) => {
            return (0, jsnetworkx_1.hasPath)(this.graph, {
                source: this._maybeHexify(this.id),
                target: id,
            });
        });
        const offline = this.graph.nodes().filter((id) => {
            return !(0, jsnetworkx_1.hasPath)(this.graph, {
                source: this._maybeHexify(this.id),
                target: id,
            });
        });
        for (const id of offline) {
            this.graph.removeNode(id);
        }
        this._online = online;
        this.emit("online", online);
    }
    _maybeHexify(data) {
        if (Buffer.isBuffer(data)) {
            return data.toString("hex");
        }
        return data;
    }
}
exports.default = DHTOnlineBase;