Kieran/snort
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: relay-worker insert replacable events

Browse Source
This commit is contained in:
Kieran 2024-01-18 16:00:20 +00:00
parent e3f8d48ddb
commit f147edd03c
No known key found for this signature in database
GPG Key ID: DE71CEB3925BE941
8 changed files with 155 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
packages/app/src/Feed/LoginFeed.ts
View File

@ -41,11 +41,7 @@ export default function useLoginFeed() {
const { publicKey: pubKey, follows } = login;
const { publisher, system } = useEventPublisher();
const followLists = useFollowsContactListView();
useEffect(() => {
followLists.forEach(e => socialGraphInstance.handleEvent(e));
}, followLists);
useFollowsContactListView();
useEffect(() => {
system.checkSigs = login.appData.item.preferences.checkSigs;
}, [login]);

6
packages/app/src/Feed/WorkerRelayView.ts
View File

@ -36,13 +36,13 @@ export function useWorkerRelayView(id: string, filters: Array<ReqFilter>, leaveO
...f,
limit: undefined,
until: undefined,
since: latest.results?.at(i)?.created_at ?? (maxWindow ? unixNow() - maxWindow : undefined),
since: latest.result?.at(i)?.created_at ?? (maxWindow ? unixNow() - maxWindow : undefined),
}))
.forEach(f => rb.withBareFilter(f));
setRb(rb);
});
Relay.req({ id, filters, leaveOpen }).then(res => {
setEvents(res.results);
setEvents(res.result);
if (res.port) {
res.port.addEventListener("message", ev => {
const evs = ev.data as Array<NostrEvent>;
@ -82,7 +82,7 @@ export function useWorkerRelayViewCount(id: string, filters: Array<ReqFilter>, m
...f,
limit: undefined,
until: undefined,
since: latest.results?.at(i)?.created_at ?? (maxWindow ? unixNow() - maxWindow : undefined),
since: latest.result?.at(i)?.created_at ?? (maxWindow ? unixNow() - maxWindow : undefined),
}))
.forEach(f => rb.withBareFilter(f));
setRb(rb);

38
packages/app/src/Pages/settings/Cache.tsx
View File

@ -74,26 +74,34 @@ function RelayCacheStats() {
</tr>
</thead>
<tbody>
{Object.entries(counts)
.sort(([, a], [, b]) => (a > b ? -1 : 1))
.map(([k, v]) => {
return (
<tr key={k}>
<td>{k}</td>
<td>
<FormattedNumber value={v} />
</td>
</tr>
);
})}
{Object.entries(counts).sort(([, a], [, b]) => a > b ? -1 : 1).map(([k, v]) => {
return (
<tr key={k}>
<td><FormattedNumber value={Number(k)} /></td>
<td><FormattedNumber value={v} /></td>
</tr>
);
})}
</tbody>
</table>
</div>
<div>
<AsyncButton onClick={() => {}}>
<div className="flex flex-col gap-2">
<AsyncButton onClick={() => { }}>
<FormattedMessage defaultMessage="Clear" id="/GCoTA" />
</AsyncButton>
<AsyncButton onClick={async () => {
const data = await Relay.dump();
const url = URL.createObjectURL(new File([data], "snort.db", {
type: "application/octet-stream"
}));
const a = document.createElement("a");
a.href = url;
a.download = "snort.db";
a.click();
}}>
<FormattedMessage defaultMessage="Dump" id="f2CAxA" />
</AsyncButton>
</div>
</div>
</div >
);
}

20
packages/app/src/system.ts
View File

@ -1,5 +1,4 @@
import { removeUndefined, throwIfOffline } from "@snort/shared";
import LRUSet from "@snort/shared/src/LRUSet";
import { mapEventToProfile, NostrEvent, NostrSystem, ProfileLoaderService, socialGraphInstance } from "@snort/system";
import { WorkerRelayInterface } from "@snort/worker-relay";
import WorkerRelayPath from "@snort/worker-relay/dist/worker?worker&url";
@ -63,28 +62,13 @@ export async function fetchProfile(key: string) {
}
export const Relay = new WorkerRelayInterface(WorkerRelayPath);
let relayInitStarted = false;
export async function initRelayWorker() {
if (relayInitStarted) return;
relayInitStarted = true;
try {
if (await Relay.init()) {
if (await Relay.open()) {
await Relay.migrate();
const seen = new LRUSet<string>(100);
System.on("event", async (_, ev) => {
if (seen.has(ev.id)) return;
seen.add(ev.id);
await Relay.event(ev);
});
System.on("request", async (subId, f) => {
const evs = await Relay.req(["REQ", "", ...f.filters]);
evs.forEach(ev => {
seen.add(ev.id);
queueMicrotask(() => {
System.HandleEvent(subId, { ...ev, relays: [] });
});
});
System.on("event", (_, ev) => {
Relay.event(ev);
});
}
}

32
packages/worker-relay/src/interface.ts
View File

@ -21,35 +21,39 @@ export class WorkerRelayInterface {
}
async init() {
return await this.#workerRpc<void, boolean>("init");
return (await this.#workerRpc<void, boolean>("init")).result;
}
async open() {
return await this.#workerRpc<void, boolean>("open");
return (await this.#workerRpc<void, boolean>("open")).result;
}
async migrate() {
return await this.#workerRpc<void, boolean>("migrate");
return (await this.#workerRpc<void, boolean>("migrate")).result;
}
async event(ev: NostrEvent) {
return await this.#workerRpc<NostrEvent, boolean>("event", ev);
return (await this.#workerRpc<NostrEvent, boolean>("event", ev)).result;
}
async req(req: ReqCommand) {
return await this.#workerRpc<ReqCommand, { results: Array<NostrEvent>; port?: Readonly<MessagePort> }>("req", req);
return await this.#workerRpc<ReqCommand, Array<NostrEvent>>("req", req);
}
async count(req: ReqCommand) {
return await this.#workerRpc<ReqCommand, number>("count", req);
return (await this.#workerRpc<ReqCommand, number>("count", req)).result;
}
async summary() {
return await this.#workerRpc<void, Record<string, number>>("summary");
return (await this.#workerRpc<void, Record<string, number>>("summary")).result;
}
async close(id: string) {
return await this.#workerRpc<string, boolean>("close", id);
return (await this.#workerRpc<string, boolean>("close", id)).result;
}
async dump() {
return (await this.#workerRpc<void, Uint8Array>("dumpDb")).result;
}
#workerRpc<T, R>(cmd: string, args?: T, timeout = 30_000) {
@ -60,12 +64,18 @@ export class WorkerRelayInterface {
args,
} as WorkerMessage<T>;
this.#worker.postMessage(msg);
return new Promise<R>((resolve, reject) => {
return new Promise<{
result: R;
port: MessagePort | undefined;
}>((resolve, reject) => {
let t: ReturnType<typeof setTimeout>;
this.#commandQueue.set(id, (v, ports) => {
this.#commandQueue.set(id, (v, port) => {
clearTimeout(t);
const cmdReply = v as WorkerMessage<R>;
resolve({ ...cmdReply.args, port: ports.length > 0 ? ports[0] : undefined });
resolve({
result: cmdReply.args,
port: port.length > 0 ? port[0] : undefined,
});
});
t = setTimeout(() => {
reject("timeout");

128
packages/worker-relay/src/relay.ts
View File

@ -9,8 +9,9 @@ export interface WorkerRelayEvents {
export class WorkerRelay extends EventEmitter<WorkerRelayEvents> {
#sqlite?: Sqlite3Static;
#log = debug("WorkerRelay");
#log = (...args: any[]) => console.debug(...args);
#db?: Database;
#seenInserts = new Set<string>();
/**
* Initialize the SQLite driver
@ -101,27 +102,45 @@ export class WorkerRelay extends EventEmitter<WorkerRelayEvents> {
return eventsInserted.length > 0;
}
#deleteById(db: Database, ids: Array<string>) {
db.exec(`delete from events where id in (${this.#repeatParams(ids.length)})`, {
bind: ids,
});
}
#insertEvent(db: Database, ev: NostrEvent) {
if (this.#seenInserts.has(ev.id)) return false;
const legacyReplacable = [0, 3, 41];
if (legacyReplacable.includes(ev.kind) || (ev.kind >= 10_000 && ev.kind < 20_000)) {
db.exec("delete from events where kind = ? and pubkey = ? and created < ?", {
bind: [ev.kind, ev.pubkey, ev.created_at],
});
const oldDeleted = db.changes();
if (oldDeleted === 0) {
const oldEvents = db.selectValues("select id from events where kind = ? and pubkey = ? and created <= ?", [
ev.kind,
ev.pubkey,
ev.created_at,
]) as Array<string>;
if (oldEvents.includes(ev.id)) {
// we already have this event, return
this.#seenInserts.add(ev.id);
if (oldEvents.length > 1) {
const toDelete = oldEvents.filter(a => a !== ev.id);
this.#deleteById(db, toDelete);
}
return false;
}
}
if (ev.kind >= 30_000 && ev.kind < 40_000) {
const dTag = ev.tags.find(a => a[0] === "d")![1];
db.exec(
"delete from events where id in (select id from events, tags where events.id = tags.event_id and tags.key = ? and tags.value = ?)",
{
bind: ["d", dTag],
},
);
const oldDeleted = db.changes();
if (oldDeleted === 0) {
const oldEvents = db.selectValues(
"select id from events where id in (select id from events, tags where events.id = tags.event_id and tags.key = ? and tags.value = ?)",
["d", dTag],
) as Array<string>;
if (oldEvents.includes(ev.id)) {
// we have this version
this.#seenInserts.add(ev.id);
if (oldEvents.length > 1) {
const toDelete = oldEvents.filter(a => a !== ev.id);
this.#deleteById(db, toDelete);
}
return false;
}
}
@ -136,23 +155,21 @@ export class WorkerRelay extends EventEmitter<WorkerRelayEvents> {
});
}
}
this.#seenInserts.add(ev.id);
return eventInserted;
}
/**
* Query relay by nostr filter
*/
req(req: ReqFilter) {
req(id: string, req: ReqFilter) {
const start = unixNowMs();
const [sql, params] = this.#buildQuery(req);
const rows = this.#db?.exec(sql, {
bind: params,
returnValue: "resultRows",
});
const results = rows?.map(a => JSON.parse(a[0] as string) as NostrEvent) ?? [];
const res = this.#db?.selectArrays(sql, params);
const results = res?.map(a => JSON.parse(a[0] as string) as NostrEvent) ?? [];
const time = unixNowMs() - start;
this.#log(`Query results took ${time.toLocaleString()}ms`);
//this.#log(`Query ${id} results took ${time.toLocaleString()}ms`);
return results;
}
@ -182,38 +199,55 @@ export class WorkerRelay extends EventEmitter<WorkerRelayEvents> {
return Object.fromEntries(res?.map(a => [String(a[0]), a[1] as number]) ?? []);
}
#buildQuery(req: ReqFilter, count = false) {
/**
* Dump the database file
*/
async dump() {
const filePath = String(this.#db?.filename ?? "");
try {
this.#db?.close();
this.#db = undefined;
const dir = await navigator.storage.getDirectory();
// @ts-expect-error
for await (const [name, file] of dir) {
if (`/${name}` === filePath) {
const fh = await (file as FileSystemFileHandle).getFile();
const ret = new Uint8Array(await fh.arrayBuffer());
return ret;
}
}
} catch (e) {
console.error(e);
} finally {
this.open(filePath);
}
return new Uint8Array();
}
#buildQuery(req: ReqFilter, count = false): [string, Array<any>] {
const conditions: Array<string> = [];
const params: Array<any> = [];
const repeatParams = (n: number) => {
const ret: Array<string> = [];
for (let x = 0; x < n; x++) {
ret.push("?");
}
return ret.join(", ");
};
let sql = `select ${count ? "count(json)" : "json"} from events`;
const tags = Object.entries(req).filter(([k]) => k.startsWith("#"));
for (const [key, values] of tags) {
const vArray = values as Array<string>;
sql += ` inner join tags on events.id = tags.event_id and tags.key = ? and tags.value in (${repeatParams(
sql += ` inner join tags on events.id = tags.event_id and tags.key = ? and tags.value in (${this.#repeatParams(
vArray.length,
)})`;
params.push(key.slice(1));
params.push(...vArray);
}
if (req.ids) {
conditions.push(`id in (${repeatParams(req.ids.length)})`);
conditions.push(`id in (${this.#repeatParams(req.ids.length)})`);
params.push(...req.ids);
}
if (req.authors) {
conditions.push(`pubkey in (${repeatParams(req.authors.length)})`);
conditions.push(`pubkey in (${this.#repeatParams(req.authors.length)})`);
params.push(...req.authors);
}
if (req.kinds) {
conditions.push(`kind in (${repeatParams(req.kinds.length)})`);
conditions.push(`kind in (${this.#repeatParams(req.kinds.length)})`);
params.push(...req.kinds);
}
if (req.since) {
@ -233,6 +267,32 @@ export class WorkerRelay extends EventEmitter<WorkerRelayEvents> {
return [sql, params];
}
#repeatParams(n: number) {
const ret: Array<string> = [];
for (let x = 0; x < n; x++) {
ret.push("?");
}
return ret.join(", ");
}
#replaceParamsDebug(sql: string, params: Array<number | string>) {
let res = "";
let cIdx = 0;
for (const chr of sql) {
if (chr === "?") {
const px = params[cIdx++];
if (typeof px === "number") {
res += px.toString();
} else if (typeof px === "string") {
res += `'${px}'`;
}
} else {
res += chr;
}
}
return res;
}
#migrate_v1() {
this.#db?.transaction(db => {
db.exec(

2
packages/worker-relay/src/types.ts
View File

@ -1,6 +1,6 @@
export interface WorkerMessage<T> {
id: string;
cmd: "reply" | "init" | "open" | "migrate" | "event" | "req" | "count" | "summary" | "close";
cmd: "reply" | "init" | "open" | "migrate" | "event" | "req" | "count" | "summary" | "close" | "dumpDb";
args: T;
}

14
packages/worker-relay/src/worker.ts
View File

@ -60,8 +60,6 @@ globalThis.onclose = () => {
};
globalThis.onmessage = ev => {
//console.debug(ev);
const msg = ev.data as WorkerMessage<any>;
try {
switch (msg.cmd) {
@ -108,9 +106,9 @@ globalThis.onmessage = ev => {
}
const results = [];
for (const r of req.filters) {
results.push(...relay.req(r as ReqFilter));
results.push(...relay.req(req.id, r as ReqFilter));
}
reply(msg.id, { results }, req.leaveOpen ? [chan.port2] : undefined);
reply(msg.id, results, req.leaveOpen ? [chan.port2] : undefined);
});
break;
}
@ -133,12 +131,20 @@ globalThis.onmessage = ev => {
});
break;
}
case "dumpDb": {
barrierQueue(cmdQueue, async () => {
const res = await relay.dump();
reply(msg.id, res);
});
break;
}
default: {
reply(msg.id, { error: "Unknown command" });
break;
}
}
} catch (e) {
console.error(e);
reply(msg.id, { error: e });
}
};