LumeWeb
/
relay
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

*remove rpc cache system, as it is not needed for now. This can be reverted in the future.

This commit is contained in:
Derrick Hammer 2023-04-18 20:19:27 -04:00
parent f720f40f05
commit 7f1dde272a
Signed by: pcfreak30
GPG Key ID: C997C339BE476FF2
2 changed files with 2 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

119
src/modules/rpc/cache.ts
View File

@ -1,119 +0,0 @@
import EventEmitter from "events";
import DHTCache from "@lumeweb/dht-cache";
import { RPCCacheItem, RPCRequest, RPCResponse } from "@lumeweb/relay-types";
import { get as getSwarm } from "../swarm";
import { RPCServer } from "./server";
// @ts-ignore
import jsonStringify from "json-stringify-deterministic";
// @ts-ignore
import crypto from "hypercore-crypto";
import NodeCache from "node-cache";
import log from "../../log.js";
export class RPCCache extends EventEmitter {
private server: RPCServer;
constructor(server: RPCServer) {
super();
this.server = server;
this._swarm = getSwarm();
this._dhtCache = new DHTCache(this._swarm, {
protocol: "lumeweb.rpccache",
logger: log.child({ module: "dht-cache" }),
});
this._data.on("del", (key: string) => {
try {
this.deleteItem(key);
} catch {}
});
}
private _dhtCache?: DHTCache;
get dhtCache(): DHTCache {
return this._dhtCache as DHTCache;
}
private _swarm?: any;
get swarm(): any {
return this._swarm;
}
private _data: NodeCache = new NodeCache({ stdTTL: 60 * 60 * 24 });
get data(): NodeCache {
return this._data;
}
public signResponse(item: RPCCacheItem): string {
const field = item.value.signedField || "data";
const updated = item.value.updated;
// @ts-ignore
let json = item.value[field];
if (typeof json !== "string") {
json = jsonStringify(json);
}
return this.server.signData(`${updated}${json}`);
}
public verifyResponse(pubkey: Buffer, item: RPCCacheItem): boolean | Buffer {
const field = item.value.signedField || "data";
const updated = item.value.updated;
// @ts-ignore
let json = item.value[field];
if (typeof json !== "string") {
json = jsonStringify(json);
}
try {
if (
!crypto.verify(
Buffer.from(`${updated}${json}`),
Buffer.from(item?.signature as string, "hex"),
pubkey
)
) {
return false;
}
} catch {
return false;
}
return true;
}
public addItem(query: RPCRequest, response: RPCResponse) {
const queryHash = RPCServer.hashQuery(query);
const clonedResponse = { ...response };
clonedResponse.updated = Date.now();
const item = {
value: clonedResponse,
signature: "",
};
item.signature = this.signResponse(item);
this._dhtCache?.addItem(queryHash);
this._data.set(queryHash, item);
}
public deleteItem(queryHash: string): boolean {
const cache = this._dhtCache?.cache;
if (!cache?.includes(queryHash)) {
throw Error("item does not exist");
}
this._dhtCache?.removeItem(queryHash);
this._data.del(queryHash);
return true;
}
}

48
src/modules/rpc/server.ts
View File

@ -8,17 +8,15 @@ import EventEmitter from "events";
// @ts-ignore
import ProtomuxRPC from "protomux-rpc";
import b4a from "b4a";
import { SecretStream } from "../swarm";
import { get as getSwarm, SecretStream } from "../swarm";
// @ts-ignore
import c from "compact-encoding";
// @ts-ignore
import crypto from "hypercore-crypto";
// @ts-ignore
import { Mutex } from "async-mutex";
import { RPCCache } from "./cache";
// @ts-ignore
import jsonStringify from "json-stringify-deterministic";
import config from "../../config";
const sodium = require("sodium-universal");
let server: RPCServer;
@ -55,19 +53,6 @@ export class RPCServer extends EventEmitter {
>();
private pendingRequests: Map<string, Mutex> = new Map<string, Mutex>();
private _cache?: RPCCache;
constructor() {
super();
if (config.bool("cache")) {
this._cache = new RPCCache(this);
}
}
get cache(): RPCCache | undefined {
return this._cache;
}
public static hashQuery(query: RPCRequest): string {
const clonedQuery: RPCRequest = {
module: query.module,
@ -145,7 +130,7 @@ export class RPCServer extends EventEmitter {
}
return crypto
.sign(Buffer.from(raw), this._cache?.swarm.keyPair.secretKey)
.sign(Buffer.from(raw), getSwarm().keyPair.secretKey)
.toString("hex");
}
@ -156,13 +141,6 @@ export class RPCServer extends EventEmitter {
return lockedRequest;
}
let cachedRequest = this.getCachedRequest(request) as RPCCacheItem;
if (cachedRequest) {
this.getRequestLock(request)?.release();
return { ...cachedRequest.value, signature: cachedRequest.signature };
}
let method = this.getMethodByRequest(request);
let ret;
@ -208,30 +186,11 @@ export class RPCServer extends EventEmitter {
};
}
method = method as RPCMethod;
if (config.bool("cache") && method.cacheable) {
this.cache?.addItem(request, rpcResult);
}
this.getRequestLock(request)?.release();
return rpcResult;
}
private getCachedRequest(request: RPCRequest): RPCCacheItem | boolean {
if (!config.bool("cache")) {
return false;
}
const req = RPCServer.hashQuery(request);
if (this._cache?.data.has(req)) {
return this._cache?.data.get<RPCCacheItem>(req) as RPCCacheItem;
}
return false;
}
private getMethodByRequest(request: RPCRequest): Error | RPCMethod {
return this.getMethod(request.module, request.method);
}
@ -269,9 +228,6 @@ export class RPCServer extends EventEmitter {
if (lock.isLocked()) {
await lock.waitForUnlock();
if (this._cache?.data.has(reqId)) {
return this._cache?.data.get<RPCCacheItem>(reqId) as RPCCacheItem;
}
}
await lock.acquire();