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

refactor: reactions grouping and other fixes

Browse Source
This commit is contained in:
Kieran
2024-01-09 16:40:31 +00:00
parent 4455651d47
commit 80fa5a132b
58 changed files with 344 additions and 501 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
packages/system/README.md
View File

@ -5,12 +5,7 @@ A collection of caching and querying techniquies used by https://snort.social to
Simple example:
```js
import {
NostrSystem,
RequestBuilder,
StoreSnapshot,
NoteCollection
} from "@snort/system"
import { NostrSystem, RequestBuilder, StoreSnapshot, NoteCollection } from "@snort/system";
// Singleton instance to store all connections and access query fetching system
const System = new NostrSystem({});
@ -30,25 +25,11 @@ const System = new NostrSystem({});
.kinds([1])
.limit(10);
const q = System.Query(NoteCollection, rb);
const q = System.Query(rb);
// basic usage using "onEvent", fired every 100ms
q.feed.onEvent(evs => {
q.on("event", evs => {
console.log(evs);
// something else..
});
// Hookable type using change notification, limited to every 500ms
const release = q.feed.hook(() => {
// since we use the NoteCollection we expect NostrEvent[]
// other stores provide different data, like a single event instead of an array (latest version)
const state = q.feed.snapshot as StoreSnapshot<ReturnType<NoteCollection["getSnapshotData"]>>;
// do something with snapshot of store
console.log(`We have ${state.data?.length} events now!`);
});
// release the hook when its not needed anymore
// these patterns will be managed in @snort/system-react to make it easier to use react or other UI frameworks
release();
})();
```

18
packages/system/examples/simple.ts
View File

@ -18,24 +18,10 @@ const System = new NostrSystem({});
.kinds([1])
.limit(10);
const q = System.Query(NoteCollection, rb);
const q = System.Query(rb);
// basic usage using "onEvent", fired every 100ms
q.feed.onEvent(evs => {
q.on("event", evs => {
console.log(evs);
// something else..
});
// Hookable type using change notification, limited to every 500ms
const release = q.feed.hook(() => {
// since we use the FlatNoteStore we expect NostrEvent[]
// other stores provide different data, like a single event instead of an array (latest version)
const state = q.feed.snapshot as StoreSnapshot<ReturnType<NoteCollection["getSnapshotData"]>>;
// do something with snapshot of store
console.log(`We have ${state.data?.length} events now!`);
});
// release the hook when its not needed anymore
// these patterns will be managed in @snort/system-react to make it easier to use react or other UI frameworks
release();
})();

13
packages/system/src/background-loader.ts
View File

@ -76,16 +76,17 @@ export abstract class BackgroundLoader<T extends { loaded: number; created: numb
} else {
return await new Promise<T>((resolve, reject) => {
this.TrackKeys(key);
this.cache.on("change", keys => {
const handler = (keys: Array<string>) => {
if (keys.includes(key)) {
const existing = this.cache.getFromCache(key);
if (existing) {
resolve(existing);
this.UntrackKeys(key);
this.cache.off("change");
this.cache.off("change", handler);
}
}
});
};
this.cache.on("change", handler);
});
}
}
@ -103,6 +104,7 @@ export abstract class BackgroundLoader<T extends { loaded: number; created: numb
missing.filter(a => !found.some(b => a === this.cache.key(b))).map(a => this.makePlaceholder(a)),
);
if (noResult.length > 0) {
this.#log("Adding placeholders for %O", noResult);
await Promise.all(noResult.map(a => this.cache.update(a)));
}
} catch (e) {
@ -115,19 +117,24 @@ export abstract class BackgroundLoader<T extends { loaded: number; created: numb
}
async #loadData(missing: Array<string>) {
this.#log("Loading data", missing);
if (this.loaderFn) {
const results = await this.loaderFn(missing);
await Promise.all(results.map(a => this.cache.update(a)));
return results;
} else {
const hookHandled = new Set<string>();
const v = await this.#system.Fetch(this.buildSub(missing), async e => {
this.#log("Callback handled %o", e);
for (const pe of e) {
const m = this.onEvent(pe);
if (m) {
await this.cache.update(m);
hookHandled.add(pe.id);
}
}
});
this.#log("Got data", v);
return removeUndefined(v.map(this.onEvent));
}
}

1
packages/system/src/cache/index.ts vendored
View File

@ -50,6 +50,7 @@ export interface UsersRelays {
}
export function mapEventToProfile(ev: NostrEvent) {
if (ev.kind !== 0) return;
try {
const data: UserMetadata = JSON.parse(ev.content);
let ret = {

17
packages/system/src/nostr-connection-pool.ts → packages/system/src/connection-pool.ts
View File

@ -30,7 +30,9 @@ export type ConnectionPool = {
/**
* Simple connection pool containing connections to multiple nostr relays
*/
export class NostrConnectionPool extends EventEmitter<NostrConnectionPoolEvents> implements ConnectionPool {
export class DefaultConnectionPool extends EventEmitter<NostrConnectionPoolEvents> implements ConnectionPool {
#system: SystemInterface;
#log = debug("NostrConnectionPool");
/**
@ -38,6 +40,11 @@ export class NostrConnectionPool extends EventEmitter<NostrConnectionPoolEvents>
*/
#sockets = new Map<string, Connection>();
constructor(system: SystemInterface) {
super();
this.#system = system;
}
/**
* Get basic state information from the pool
*/
@ -63,7 +70,13 @@ export class NostrConnectionPool extends EventEmitter<NostrConnectionPoolEvents>
const c = new Connection(addr, options, ephemeral);
this.#sockets.set(addr, c);
c.on("event", (s, e) => this.emit("event", addr, s, e));
c.on("event", (s, e) => {
if (this.#system.checkSigs && !this.#system.optimizer.schnorrVerify(e)) {
this.#log("Reject invalid event %o", e);
return;
}
this.emit("event", addr, s, e);
});
c.on("eose", s => this.emit("eose", addr, s));
c.on("disconnect", code => this.emit("disconnect", addr, code));
c.on("connected", r => this.emit("connected", addr, r));

11
packages/system/src/connection.ts
View File

@ -9,6 +9,7 @@ import { ConnectionStats } from "./connection-stats";
import { NostrEvent, ReqCommand, ReqFilter, TaggedNostrEvent, u256 } from "./nostr";
import { RelayInfo } from "./relay-info";
import EventKind from "./event-kind";
import { EventExt } from "./event-ext";
/**
* Relay settings
@ -225,10 +226,16 @@ export class Connection extends EventEmitter<ConnectionEvents> {
break;
}
case "EVENT": {
this.emit("event", msg[1] as string, {
const ev = {
...(msg[2] as NostrEvent),
relays: [this.Address],
});
} as TaggedNostrEvent;
if (!EventExt.isValid(ev)) {
//this.#log("Rejecting invalid event %O", ev);
return;
}
this.emit("event", msg[1] as string, ev);
this.Stats.EventsReceived++;
this.notifyChange();
break;

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

@ -1,13 +1,14 @@
import { RelaySettings, ConnectionStateSnapshot, OkResponse } from "./connection";
import { RequestBuilder } from "./request-builder";
import { NoteCollection, NoteStore, NoteStoreSnapshotData, StoreSnapshot } from "./note-collection";
import { NostrEvent, ReqFilter, TaggedNostrEvent } from "./nostr";
import { ProfileLoaderService } from "./profile-cache";
import { RelayCache, RelayMetadataLoader } from "./outbox-model";
import { Optimizer } from "./query-optimizer";
import { base64 } from "@scure/base";
import { FeedCache } from "@snort/shared";
import { ConnectionPool } from "nostr-connection-pool";
import { ConnectionPool } from "./connection-pool";
import EventEmitter from "eventemitter3";
import { QueryEvents } from "./query";
export { NostrSystem } from "./nostr-system";
export { default as EventKind } from "./event-kind";
@ -46,13 +47,16 @@ export * from "./cache/relay-metric";
export * from "./worker/system-worker";
export interface QueryLike {
on: (event: "event", fn?: (evs: Array<TaggedNostrEvent>) => void) => void;
off: (event: "event", fn?: (evs: Array<TaggedNostrEvent>) => void) => void;
export type QueryLike = {
get progress(): number;
feed: {
add: (evs: Array<TaggedNostrEvent>) => void;
clear: () => void;
};
cancel: () => void;
uncancel: () => void;
get snapshot(): StoreSnapshot<Array<TaggedNostrEvent>>;
}
get snapshot(): Array<TaggedNostrEvent>;
} & EventEmitter<QueryEvents>;
export interface SystemInterface {
/**
@ -87,7 +91,7 @@ export interface SystemInterface {
* @param req Request to send to relays
* @param cb A callback which will fire every 100ms when new data is received
*/
Fetch(req: RequestBuilder, cb?: (evs: ReadonlyArray<TaggedNostrEvent>) => void): Promise<Array<TaggedNostrEvent>>;
Fetch(req: RequestBuilder, cb?: (evs: Array<TaggedNostrEvent>) => void): Promise<Array<TaggedNostrEvent>>;
/**
* Create a new permanent connection to a relay

40
packages/system/src/nostr-system.ts
View File

@ -23,8 +23,8 @@ import {
import { EventsCache } from "./cache/events";
import { RelayMetadataLoader } from "./outbox-model";
import { Optimizer, DefaultOptimizer } from "./query-optimizer";
import { NostrConnectionPool } from "./nostr-connection-pool";
import { NostrQueryManager } from "./nostr-query-manager";
import { ConnectionPool, DefaultConnectionPool } from "./connection-pool";
import { QueryManager } from "./query-manager";
export interface NostrSystemEvents {
change: (state: SystemSnapshot) => void;
@ -48,7 +48,7 @@ export interface NostrsystemProps {
*/
export class NostrSystem extends EventEmitter<NostrSystemEvents> implements SystemInterface {
#log = debug("System");
#queryManager: NostrQueryManager;
#queryManager: QueryManager;
/**
* Storage class for user relay lists
@ -80,7 +80,7 @@ export class NostrSystem extends EventEmitter<NostrSystemEvents> implements Syst
*/
readonly optimizer: Optimizer;
readonly pool = new NostrConnectionPool();
readonly pool: ConnectionPool;
readonly eventsCache: FeedCache<NostrEvent>;
readonly relayLoader: RelayMetadataLoader;
@ -102,7 +102,8 @@ export class NostrSystem extends EventEmitter<NostrSystemEvents> implements Syst
this.relayLoader = new RelayMetadataLoader(this, this.relayCache);
this.checkSigs = props.checkSigs ?? true;
this.#queryManager = new NostrQueryManager(this);
this.pool = new DefaultConnectionPool(this);
this.#queryManager = new QueryManager(this);
// hook connection pool
this.pool.on("connected", (id, wasReconnect) => {
@ -121,18 +122,6 @@ export class NostrSystem extends EventEmitter<NostrSystemEvents> implements Syst
});
this.pool.on("event", (_, sub, ev) => {
ev.relays?.length && this.relayMetricsHandler.onEvent(ev.relays[0]);
if (!EventExt.isValid(ev)) {
this.#log("Rejecting invalid event %O", ev);
return;
}
if (this.checkSigs) {
if (!this.optimizer.schnorrVerify(ev)) {
this.#log("Invalid sig %O", ev);
return;
}
}
this.emit("event", sub, ev);
});
this.pool.on("disconnect", (id, code) => {
@ -160,17 +149,6 @@ export class NostrSystem extends EventEmitter<NostrSystemEvents> implements Syst
this.#queryManager.on("trace", t => {
this.relayMetricsHandler.onTraceReport(t);
});
// internal handler for on-event
this.on("event", (sub, ev) => {
for (const [, v] of this.#queryManager) {
const trace = v.handleEvent(sub, ev);
// inject events to cache if query by id
if (trace && trace.filters.some(a => a.ids)) {
this.eventsCache.set(ev);
}
}
});
}
get Sockets(): ConnectionStateSnapshot[] {
@ -200,15 +178,15 @@ export class NostrSystem extends EventEmitter<NostrSystemEvents> implements Syst
}
GetQuery(id: string): QueryLike | undefined {
return this.#queryManager.get(id) as QueryLike;
return this.#queryManager.get(id);
}
Fetch(req: RequestBuilder, cb?: (evs: ReadonlyArray<TaggedNostrEvent>) => void) {
Fetch(req: RequestBuilder, cb?: (evs: Array<TaggedNostrEvent>) => void) {
return this.#queryManager.fetch(req, cb);
}
Query(req: RequestBuilder): QueryLike {
return this.#queryManager.query(req) as QueryLike;
return this.#queryManager.query(req);
}
HandleEvent(ev: TaggedNostrEvent) {

178
packages/system/src/note-collection.ts
View File

@ -1,27 +1,10 @@
import { appendDedupe, SortedMap } from "@snort/shared";
import { EventExt, EventType, TaggedNostrEvent, u256 } from ".";
import { SortedMap } from "@snort/shared";
import { EventExt, EventType, TaggedNostrEvent } from ".";
import { findTag } from "./utils";
import EventEmitter from "eventemitter3";
export interface StoreSnapshot<TSnapshot extends NoteStoreSnapshotData> {
data: TSnapshot | undefined;
clear: () => void;
loading: () => boolean;
add: (ev: Readonly<TaggedNostrEvent> | Readonly<Array<TaggedNostrEvent>>) => void;
}
export const EmptySnapshot = {
data: undefined,
clear: () => {
// empty
},
loading: () => true,
add: () => {
// empty
},
} as StoreSnapshot<Array<TaggedNostrEvent>>;
export type NoteStoreSnapshotData = Array<TaggedNostrEvent> | TaggedNostrEvent;
export const EmptySnapshot: NoteStoreSnapshotData = [];
export type NoteStoreSnapshotData = Array<TaggedNostrEvent>;
export type NoteStoreHook = () => void;
export type NoteStoreHookRelease = () => void;
export type OnEventCallback = (e: Readonly<Array<TaggedNostrEvent>>) => void;
@ -30,8 +13,7 @@ export type OnEoseCallback = (c: string) => void;
export type OnEoseCallbackRelease = () => void;
export interface NosteStoreEvents {
progress: (loading: boolean) => void;
event: (evs: Readonly<Array<TaggedNostrEvent>>) => void;
event: (evs: Array<TaggedNostrEvent>) => void;
}
/**
@ -40,133 +22,40 @@ export interface NosteStoreEvents {
export abstract class NoteStore extends EventEmitter<NosteStoreEvents> {
abstract add(ev: Readonly<TaggedNostrEvent> | Readonly<Array<TaggedNostrEvent>>): void;
abstract clear(): void;
abstract getSnapshotData(): NoteStoreSnapshotData | undefined;
abstract get snapshot(): StoreSnapshot<NoteStoreSnapshotData>;
abstract get loading(): boolean;
abstract set loading(v: boolean);
abstract get snapshot(): NoteStoreSnapshotData;
}
export abstract class HookedNoteStore<TSnapshot extends NoteStoreSnapshotData> extends NoteStore {
#loading = true;
#storeSnapshot: StoreSnapshot<TSnapshot> = {
clear: () => this.clear(),
loading: () => this.loading,
add: ev => this.add(ev),
data: undefined,
};
#needsSnapshot = true;
#nextNotifyTimer?: ReturnType<typeof setTimeout>;
export abstract class HookedNoteStore extends NoteStore {
#storeSnapshot: NoteStoreSnapshotData = [];
#nextEmit?: ReturnType<typeof setTimeout>;
#bufEmit: Array<TaggedNostrEvent> = [];
get snapshot() {
this.#updateSnapshot();
return this.#storeSnapshot;
}
get loading() {
return this.#loading;
}
set loading(v: boolean) {
this.#loading = v;
this.emit("progress", v);
}
abstract override add(ev: Readonly<TaggedNostrEvent> | Readonly<Array<TaggedNostrEvent>>): void;
abstract override clear(): void;
protected abstract takeSnapshot(): NoteStoreSnapshotData | undefined;
getSnapshotData() {
this.#updateSnapshot();
return this.#storeSnapshot.data;
}
protected abstract takeSnapshot(): TSnapshot | undefined;
protected onChange(changes: Readonly<Array<TaggedNostrEvent>>): void {
this.#needsSnapshot = true;
protected onChange(changes: Array<TaggedNostrEvent>): void {
this.#storeSnapshot = this.takeSnapshot() ?? [];
this.#bufEmit.push(...changes);
if (!this.#nextNotifyTimer) {
this.#nextNotifyTimer = setTimeout(() => {
this.#nextNotifyTimer = undefined;
if (!this.#nextEmit) {
this.#nextEmit = setTimeout(() => {
this.#nextEmit = undefined;
this.emit("event", this.#bufEmit);
this.#bufEmit = [];
}, 500);
}, 300);
}
}
#updateSnapshot() {
if (this.#needsSnapshot) {
this.#storeSnapshot = {
...this.#storeSnapshot,
data: this.takeSnapshot(),
};
this.#needsSnapshot = false;
}
}
}
/**
* A store which doesnt store anything, useful for hooks only
*/
export class NoopStore extends HookedNoteStore<Array<TaggedNostrEvent>> {
override add(ev: readonly TaggedNostrEvent[] | Readonly<TaggedNostrEvent>): void {
this.onChange(Array.isArray(ev) ? ev : [ev]);
}
override clear(): void {
// nothing to do
}
protected override takeSnapshot(): TaggedNostrEvent[] | undefined {
// nothing to do
return undefined;
}
}
/**
* A simple flat container of events with no duplicates
*/
export class FlatNoteStore extends HookedNoteStore<Array<TaggedNostrEvent>> {
#events: Array<TaggedNostrEvent> = [];
#ids: Set<u256> = new Set();
add(ev: TaggedNostrEvent | Array<TaggedNostrEvent>) {
ev = Array.isArray(ev) ? ev : [ev];
const changes: Array<TaggedNostrEvent> = [];
ev.forEach(a => {
if (!this.#ids.has(a.id)) {
this.#events.push(a);
this.#ids.add(a.id);
changes.push(a);
} else {
const existing = this.#events.findIndex(b => b.id === a.id);
if (existing !== -1) {
this.#events[existing].relays = appendDedupe(this.#events[existing].relays, a.relays);
}
}
});
if (changes.length > 0) {
this.onChange(changes);
}
}
clear() {
this.#events = [];
this.#ids.clear();
this.onChange([]);
}
takeSnapshot() {
return [...this.#events];
}
}
/**
* A note store that holds a single replaceable event for a given user defined key generator function
*/
export class KeyedReplaceableNoteStore extends HookedNoteStore<Array<TaggedNostrEvent>> {
export class KeyedReplaceableNoteStore extends HookedNoteStore {
#keyFn: (ev: TaggedNostrEvent) => string;
#events: SortedMap<string, TaggedNostrEvent> = new SortedMap([], (a, b) => b[1].created_at - a[1].created_at);
@ -201,39 +90,6 @@ export class KeyedReplaceableNoteStore extends HookedNoteStore<Array<TaggedNostr
}
}
/**
* A note store that holds a single replaceable event
*/
export class ReplaceableNoteStore extends HookedNoteStore<Readonly<TaggedNostrEvent>> {
#event?: TaggedNostrEvent;
add(ev: TaggedNostrEvent | Array<TaggedNostrEvent>) {
ev = Array.isArray(ev) ? ev : [ev];
const changes: Array<TaggedNostrEvent> = [];
ev.forEach(a => {
const existingCreated = this.#event?.created_at ?? 0;
if (a.created_at > existingCreated) {
this.#event = a;
changes.push(a);
}
});
if (changes.length > 0) {
this.onChange(changes);
}
}
clear() {
this.#event = undefined;
this.onChange([]);
}
takeSnapshot() {
if (this.#event) {
return { ...this.#event };
}
}
}
/**
* General use note store based on kind ranges
*/

11
packages/system/src/profile-cache.ts
View File

@ -2,7 +2,6 @@ import { unixNowMs } from "@snort/shared";
import { EventKind, TaggedNostrEvent, RequestBuilder } from ".";
import { ProfileCacheExpire } from "./const";
import { mapEventToProfile, CachedMetadata } from "./cache";
import { v4 as uuid } from "uuid";
import { BackgroundLoader } from "./background-loader";
export class ProfileLoaderService extends BackgroundLoader<CachedMetadata> {
@ -19,14 +18,8 @@ export class ProfileLoaderService extends BackgroundLoader<CachedMetadata> {
}
override buildSub(missing: string[]): RequestBuilder {
const sub = new RequestBuilder(`profiles-${uuid()}`);
sub
.withOptions({
skipDiff: true,
})
.withFilter()
.kinds([EventKind.SetMetadata])
.authors(missing);
const sub = new RequestBuilder(`profiles`);
sub.withFilter().kinds([EventKind.SetMetadata]).authors(missing);
return sub;
}

40
packages/system/src/nostr-query-manager.ts → packages/system/src/query-manager.ts
View File

@ -2,11 +2,10 @@ import debug from "debug";
import EventEmitter from "eventemitter3";
import { BuiltRawReqFilter, RequestBuilder, SystemInterface, TaggedNostrEvent } from ".";
import { Query, TraceReport } from "./query";
import { unwrap } from "@snort/shared";
import { FilterCacheLayer, IdsFilterCacheLayer } from "./filter-cache-layer";
import { trimFilters } from "./request-trim";
interface NostrQueryManagerEvents {
interface QueryManagerEvents {
change: () => void;
trace: (report: TraceReport) => void;
}
@ -14,8 +13,8 @@ interface NostrQueryManagerEvents {
/**
* Query manager handles sending requests to the nostr network
*/
export class NostrQueryManager extends EventEmitter<NostrQueryManagerEvents> {
#log = debug("NostrQueryManager");
export class QueryManager extends EventEmitter<QueryManagerEvents> {
#log = debug("QueryManager");
/**
* All active queries
@ -70,20 +69,27 @@ export class NostrQueryManager extends EventEmitter<NostrQueryManagerEvents> {
/**
* Async fetch results
*/
fetch(req: RequestBuilder, cb?: (evs: ReadonlyArray<TaggedNostrEvent>) => void) {
const q = this.query(req);
return new Promise<Array<TaggedNostrEvent>>(resolve => {
if (cb) {
q.feed.on("event", cb);
}
q.feed.on("progress", loading => {
async fetch(req: RequestBuilder, cb?: (evs: Array<TaggedNostrEvent>) => void) {
const q = new Query(this.#system, req);
q.on("trace", r => this.emit("trace", r));
q.on("filters", fx => {
this.#send(q, fx);
});
if (cb) {
q.on("event", evs => cb(evs));
}
await new Promise<void>(resolve => {
q.on("loading", loading => {
this.#log("loading %s %o", q.id, loading);
if (!loading) {
q.feed.off("event");
q.cancel();
resolve(unwrap(q.snapshot.data));
resolve();
}
});
});
const results = q.feed.takeSnapshot();
q.cleanup();
this.#log("Fetch results for %s %o", q.id, results);
return results;
}
*[Symbol.iterator]() {
@ -105,12 +111,13 @@ export class NostrQueryManager extends EventEmitter<NostrQueryManagerEvents> {
// check for empty filters
const fNew = trimFilters(qSend.filters);
if (fNew.length === 0) {
this.#log("Dropping %s %o", q.id, qSend);
return;
}
qSend.filters = fNew;
if (qSend.relay) {
this.#log("Sending query to %s %O", qSend.relay, qSend);
this.#log("Sending query to %s %s %O", qSend.relay, q.id, qSend);
const s = this.#system.pool.getConnection(qSend.relay);
if (s) {
const qt = q.sendToRelay(s, qSend);
@ -132,7 +139,7 @@ export class NostrQueryManager extends EventEmitter<NostrQueryManagerEvents> {
const ret = [];
for (const [a, s] of this.#system.pool) {
if (!s.Ephemeral) {
this.#log("Sending query to %s %O", a, qSend);
this.#log("Sending query to %s %s %O", a, q.id, qSend);
const qt = q.sendToRelay(s, qSend);
if (qt) {
ret.push(qt);
@ -142,6 +149,7 @@ export class NostrQueryManager extends EventEmitter<NostrQueryManagerEvents> {
return ret;
}
}
#cleanup() {
let changed = false;
for (const [k, v] of this.#queries) {

101
packages/system/src/query.ts
View File

@ -4,7 +4,7 @@ import EventEmitter from "eventemitter3";
import { unixNowMs, unwrap } from "@snort/shared";
import { Connection, ReqFilter, Nips, TaggedNostrEvent, SystemInterface } from ".";
import { NoteCollection, NoteStore } from "./note-collection";
import { NoteCollection } from "./note-collection";
import { BuiltRawReqFilter, RequestBuilder } from "./request-builder";
import { eventMatchesFilter } from "./request-matcher";
@ -97,25 +97,26 @@ export interface TraceReport {
responseTime: number;
}
interface QueryEvents {
export interface QueryEvents {
loading: (v: boolean) => void;
trace: (report: TraceReport) => void;
filters: (req: BuiltRawReqFilter) => void;
event: (evs: ReadonlyArray<TaggedNostrEvent>) => void;
event: (evs: Array<TaggedNostrEvent>) => void;
end: () => void;
}
/**
* Active or queued query on the system
*/
export class Query extends EventEmitter<QueryEvents> {
/**
* Unique id of this query
*/
readonly id: string;
get id() {
return this.request.id;
}
/**
* RequestBuilder instance
*/
requests: Array<RequestBuilder> = [];
request: RequestBuilder;
/**
* Nostr system interface
@ -166,8 +167,7 @@ export class Query extends EventEmitter<QueryEvents> {
constructor(system: SystemInterface, req: RequestBuilder) {
super();
this.id = uuid();
this.requests.push(req);
this.request = req;
this.#system = system;
this.#feed = new NoteCollection();
this.#leaveOpen = req.options?.leaveOpen ?? false;
@ -176,32 +176,21 @@ export class Query extends EventEmitter<QueryEvents> {
this.#checkTraces();
this.feed.on("event", evs => this.emit("event", evs));
this.#start();
}
/**
* Adds another request to this one
*/
addRequest(req: RequestBuilder) {
if (this.#groupTimeout) {
clearTimeout(this.#groupTimeout);
this.#groupTimeout = undefined;
}
if (this.requests.some(a => a.instance === req.instance)) {
// already exists, nothing to add
return false;
}
if (this.requests.some(a => a.options?.skipDiff !== req.options?.skipDiff)) {
throw new Error("Mixing skipDiff option is not supported");
}
this.requests.push(req);
if (this.#groupingDelay) {
this.#groupTimeout = setTimeout(() => {
this.#emitFilters();
}, this.#groupingDelay);
} else {
this.#emitFilters();
if (req.instance === this.request.instance) {
// same requst, do nothing
this.#log("Same query %O === %O", req, this.request);
return;
}
this.#log("Add query %O to %s", req, this.id);
this.request.add(req);
this.#start();
return true;
}
@ -228,12 +217,11 @@ export class Query extends EventEmitter<QueryEvents> {
return this.#feed.snapshot;
}
handleEvent(sub: string, e: TaggedNostrEvent) {
#handleEvent(sub: string, e: TaggedNostrEvent) {
for (const t of this.#tracing) {
if (t.id === sub || sub === "*") {
if (t.filters.some(v => eventMatchesFilter(e, v))) {
this.feed.add(e);
return t;
} else {
this.#log("Event did not match filter, rejecting %O %O", e, t);
}
@ -254,7 +242,12 @@ export class Query extends EventEmitter<QueryEvents> {
}
cleanup() {
if (this.#groupTimeout) {
clearTimeout(this.#groupTimeout);
this.#groupTimeout = undefined;
}
this.#stopCheckTraces();
this.emit("end");
}
/**
@ -316,35 +309,46 @@ export class Query extends EventEmitter<QueryEvents> {
return thisProgress;
}
#emitFilters() {
if (this.requests.every(a => !!a.options?.skipDiff)) {
const existing = this.filters;
const rb = new RequestBuilder(this.id);
this.requests.forEach(a => rb.add(a));
const filters = rb.buildDiff(this.#system, existing);
filters.forEach(f => this.emit("filters", f));
this.requests = [];
#start() {
if (this.#groupTimeout) {
clearTimeout(this.#groupTimeout);
this.#groupTimeout = undefined;
}
if (this.#groupingDelay) {
this.#groupTimeout = setTimeout(() => {
this.#emitFilters();
}, this.#groupingDelay);
} else {
// send without diff
const rb = new RequestBuilder(this.id);
this.requests.forEach(a => rb.add(a));
const filters = rb.build(this.#system);
this.#emitFilters();
}
}
#emitFilters() {
this.#log("Starting emit of %s", this.id);
const existing = this.filters;
if (!(this.request.options?.skipDiff ?? false) && existing.length > 0) {
const filters = this.request.buildDiff(this.#system, existing);
this.#log("Build %s %O", this.id, filters);
filters.forEach(f => this.emit("filters", f));
} else {
const filters = this.request.build(this.#system);
this.#log("Build %s %O", this.id, filters);
filters.forEach(f => this.emit("filters", f));
this.requests = [];
}
}
#onProgress() {
const isFinished = this.progress === 1;
if (this.feed.loading !== isFinished) {
this.#log("%s loading=%s, progress=%d, traces=%O", this.id, this.feed.loading, this.progress, this.#tracing);
this.feed.loading = isFinished;
if (isFinished) {
this.#log("%s loading=%s, progress=%d, traces=%O", this.id, !isFinished, this.progress, this.#tracing);
this.emit("loading", !isFinished);
}
}
#stopCheckTraces() {
if (this.#checkTrace) {
clearInterval(this.#checkTrace);
this.#checkTrace = undefined;
}
}
@ -398,6 +402,9 @@ export class Query extends EventEmitter<QueryEvents> {
responseTime: qt.responseTime,
} as TraceReport),
);
const handler = (sub: string, ev: TaggedNostrEvent) => this.#handleEvent(sub, ev);
c.on("event", handler);
this.on("end", () => c.off("event", handler));
this.#tracing.push(qt);
c.QueueReq(["REQ", qt.id, ...qt.filters], () => qt.sentToRelay());
return qt;

2
packages/system/src/worker/system-worker.ts
View File

@ -25,7 +25,7 @@ import { FeedCache } from "@snort/shared";
import { EventsCache } from "../cache/events";
import { RelayMetricHandler } from "../relay-metric-handler";
import debug from "debug";
import { ConnectionPool } from "nostr-connection-pool";
import { ConnectionPool } from "connection-pool";
export class SystemWorker extends EventEmitter<NostrSystemEvents> implements SystemInterface {
#log = debug("SystemWorker");