dht-cache/dist/index.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");
// @ts-ignore
const hypercore_crypto_1 = __importDefault(require("hypercore-crypto"));
const b4a_1 = __importDefault(require("b4a"));
const messages_js_1 = require("./messages.js");
const dht_flood_1 = __importDefault(require("@lumeweb/dht-flood"));
const EntityType = {
    PUBKEY: Symbol.for("PUBKEY"),
    ITEM: Symbol.for("ITEM"),
};
const DISCONNECT_SMOOTH = 500;
class DHTCache extends events_1.default {
    swarm;
    id;
    bootstrapped;
    graph;
    connectedTo;
    flood;
    constructor(swarm, { id = swarm.keyPair.publicKey, ...opts } = {}) {
        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._cache = new Set();
        this._online = new Set([this._maybeHexify(this.id)]);
        this.swarm = swarm;
        this.flood = new dht_flood_1.default({ id, swarm, ...opts });
        this.flood.on("peer-open", (peer) => this.addPeerHandler(peer));
        this.flood.on("peer-remove", (peer) => this.removePeerHandler(peer));
        this.flood.on("message", (message, id) => this.onGetBroadcast(message, id));
        this.swarm.on("connection", (peer) => this.flood.send(peer, b4a_1.default.from("hello"), 0));
        [...this.swarm.peers.values()]
            .map((item) => {
            remotePublicKey: item.publicKey;
        })
            .filter((item) => !!item)
            .forEach((item) => {
            this.addPeerHandler(item);
        });
        this._ensurePeer(id);
    }
    _cache;
    get cache() {
        return [...this._cache].sort();
    }
    get allCache() {
        const items = [];
        for (const id of this.graph.nodesIter()) {
            const item = this.graph.node.get(id);
            if (item?.type !== EntityType.ITEM) {
                continue;
            }
            items.push(id);
        }
        return items.sort();
    }
    peerHasItem(peer, item) {
        peer = this._maybeHexify(peer);
        item = this._maybeHexify(item);
        return this.graph.hasSuccessor(peer, item);
    }
    addItem(item) {
        item = this._maybeHexify(item);
        this._cache.add(item);
        this._ensureItem(item);
        this._addEntityConnection(this.id, item);
        const broadcast = () => {
            this._broadcastMessage({
                type: messages_js_1.Type.ADD_ITEM,
                data: b4a_1.default.from(item, "hex"),
                signature: this._signItem(item),
            });
        };
        if (!this.bootstrapped) {
            this.once("bootstrapped", broadcast);
        }
        else {
            broadcast();
        }
    }
    _compileMessage(message) {
        return messages_js_1.Message.toBinary(messages_js_1.Message.create(message));
    }
    _broadcastMessage(message, ttl) {
        this.broadcast(this._compileMessage(message), ttl);
    }
    removeItem(item) {
        item = this._maybeHexify(item);
        if (!this._cache.has(item)) {
            return false;
        }
        this._cache.delete(item);
        this._removeEntity(item);
        this._broadcastMessage({
            type: messages_js_1.Type.REMOVE_ITEM,
            data: b4a_1.default.from(item, "hex"),
            signature: this._signItem(item),
        });
        return true;
    }
    _online;
    get online() {
        return this._online;
    }
    broadcast(message, ttl) {
        this.flood.broadcast(message, ttl);
    }
    send(message) {
        this.flood.send(message, 0);
    }
    addPeerHandler(peer) {
        const id = peer.remotePublicKey;
        const stringId = id.toString("hex");
        if (this.connectedTo.has(stringId)) {
            return;
        } // Already know we're connected here
        this.connectedTo.add(stringId);
        this._ensurePeer(id);
        this._addEntityConnection(this.id, id);
        this.emit("peer-add", id);
        this._recalculate();
        this._broadcastMessage({
            type: messages_js_1.Type.CONNECTED,
            id,
        });
        if (this.bootstrapped) {
            return;
        }
        // If this is the first person we've met, get their graph
        this._broadcastMessage({ type: messages_js_1.Type.BOOTSTRAP_REQUEST }, 0);
    }
    removePeerHandler(peer) {
        const id = peer.remotePublicKey;
        // Wait for a bit and check if we're still disconnected before removing the peer
        setTimeout(() => {
            if (this.swarm._allConnections.has(id)) {
                return;
            }
            this.onRemovePeer(peer);
        }, DISCONNECT_SMOOTH);
    }
    onRemovePeer(peer) {
        const id = peer.remotePublicKey;
        this.connectedTo.delete(id.toString("hex"));
        this._removeEntityConnection(this.id, id);
        this.emit("peer-remove");
        this._recalculate();
        this._broadcastMessage({
            type: messages_js_1.Type.DISCONNECTED,
            id,
        }, 0);
    }
    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 ([messages_js_1.Type.ADD_ITEM, messages_js_1.Type.REMOVE_ITEM].includes(type)) {
            const { data: rawData, signature } = decoded;
            const bufData = b4a_1.default.from(rawData);
            if (signature && hypercore_crypto_1.default.verify(bufData, signature, id)) {
                if (messages_js_1.Type.ADD_ITEM === type) {
                    this._ensureItem(bufData);
                    this._addEntityConnection(id, bufData);
                    this.emit("item-added", id, bufData);
                }
                if (messages_js_1.Type.REMOVE_ITEM === type) {
                    this.removeItem(bufData);
                    this._pruneItems();
                    this.emit("item-removed", id, bufData);
                }
            }
        }
        else if (type === messages_js_1.Type.CONNECTED) {
            const { id: toId } = decoded;
            this._addEntityConnection(id, b4a_1.default.from(toId));
            this.emit("peer-add-seen", id, toId);
            this._recalculate();
        }
        else if (type === messages_js_1.Type.DISCONNECTED) {
            const { id: toId } = decoded;
            this._removeEntityConnection(id, b4a_1.default.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);
        }
    }
    _signItem(item) {
        item = this._maybeHexify(item);
        return hypercore_crypto_1.default.sign(b4a_1.default.from(item, "hex"), this.swarm.keyPair.secretKey);
    }
    _setEntity(id, data) {
        this.graph.addNode(this._maybeHexify(id), data);
    }
    _ensureItem(id) {
        this._ensureEntity(id, { type: EntityType.ITEM });
    }
    _ensurePeer(id) {
        this._ensureEntity(id, { type: EntityType.PUBKEY });
    }
    _ensureEntity(id, def = {}) {
        id = this._maybeHexify(id);
        if (!this._hasSeenEntity(id)) {
            this._setEntity(id, def);
        }
    }
    _hasSeenEntity(id) {
        return this.graph.hasNode(this._maybeHexify(id));
    }
    _addEntityConnection(origin, destination) {
        this._ensureEntity(origin);
        this._ensureEntity(destination);
        this.graph.addEdge(this._maybeHexify(origin), this._maybeHexify(destination));
    }
    _removeEntityConnection(origin, destination) {
        try {
            this._ensureEntity(origin);
            this._ensureEntity(destination);
            this.graph.removeEdge(origin.toString("hex"), destination.toString("hex"));
        }
        catch (e) {
            if (e.name !== "JSNetworkXError")
                throw e;
        }
    }
    _removeEntity(id) {
        this.graph.removeNode(this._maybeHexify(id));
    }
    _bootstrapFrom(bootstrap) {
        if (this.bootstrapped) {
            return;
        }
        for (const id in bootstrap) {
            const { connectedTo } = bootstrap[id];
            if (id === this.id.toString("hex")) {
                continue;
            }
            for (const connection of connectedTo) {
                const peer = b4a_1.default.from(connection);
                this._ensurePeer(peer);
                this._addEntityConnection(id, peer);
            }
        }
        this.emit("bootstrapped");
        this._recalculate();
    }
    _getBootstrapInfo() {
        const state = {};
        for (const id of this.graph.nodesIter()) {
            const item = this.graph.node.get(id);
            if (item?.type !== EntityType.PUBKEY) {
                continue;
            }
            const connectedTo = this.graph
                .neighbors(id)
                .filter((item) => this.graph.node.get(item)?.type === EntityType.PUBKEY)
                .map((id) => b4a_1.default.from(id, "hex"));
            state[id] = { connectedTo };
        }
        return state;
    }
    // Calculate who's online and emit an event
    _recalculate() {
        const online = new Set();
        const offline = new Set();
        for (const id of this.graph.nodesIter()) {
            const item = this.graph.node.get(id);
            if (item?.type !== EntityType.PUBKEY) {
                continue;
            }
            if ((0, jsnetworkx_1.hasPath)(this.graph, {
                source: this._maybeHexify(this.id),
                target: id,
            })) {
                online.add(id);
            }
            else {
                offline.add(id);
            }
        }
        for (const id of offline) {
            this.graph.removeNode(id);
        }
        this._online = online;
        this.emit("online", online);
    }
    _maybeHexify(data) {
        if (b4a_1.default.isBuffer(data)) {
            return data.toString("hex");
        }
        return data;
    }
    _pruneItems(item) {
        let items;
        if (!item) {
            items = this.graph.nodesIter();
        }
        else {
            items = [item];
        }
        (async () => {
            for (const id of items) {
                const item = this.graph.node.get(id);
                if (item?.type !== EntityType.ITEM) {
                    continue;
                }
                (async () => {
                    try {
                        if (0 === this.graph.neighbors(id)?.length) {
                            this.graph.removeNode(id);
                        }
                    }
                    catch { }
                })();
            }
        })();
    }
}
exports.default = DHTCache;