vivganes/snort
1
0
Fork 0
forked from Kieran/snort
Code Pull Requests Activity

feat: event emitter

Browse Source
This commit is contained in:
Kieran 2023-11-07 13:28:01 +00:00
parent fc3d196f48
commit fcd2c8a3a0
Signed by: Kieran
GPG Key ID: DE71CEB3925BE941
8 changed files with 87 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
packages/app/src/Wallet/NostrWalletConnect.ts
View File

@ -80,15 +80,16 @@ export class NostrConnectWallet implements LNWallet {
return await new Promise<boolean>(resolve => {
this.#conn = new Connection(this.#config.relayUrl, { read: true, write: true });
this.#conn.OnConnected = () => resolve(true);
this.#conn.Auth = async (c, r) => {
this.#conn.on("connected", () => resolve(true));
this.#conn.on("auth", async (c, r, cb) => {
const eb = new EventBuilder();
eb.kind(EventKind.Auth).tag(["relay", r]).tag(["challenge", c]);
return await eb.buildAndSign(this.#config.secret);
};
this.#conn.OnEvent = (s, e) => {
const ev = await eb.buildAndSign(this.#config.secret);
cb(ev);
});
this.#conn.on("event", (s, e) => {
this.#onReply(s, e);
};
});
this.#conn.Connect();
});
}

15
packages/app/src/index.tsx
View File

@ -87,13 +87,14 @@ const System = new NostrSystem({
relayMetrics: RelayMetrics,
queryOptimizer: hasWasm ? WasmQueryOptimizer : undefined,
db: SystemDb,
authHandler: async (c, r) => {
const { id } = LoginStore.snapshot();
const pub = LoginStore.getPublisher(id);
if (pub) {
return await pub.nip42Auth(c, r);
}
},
});
System.on("auth", async (c, r, cb) => {
const { id } = LoginStore.snapshot();
const pub = LoginStore.getPublisher(id);
if (pub) {
cb(await pub.nip42Auth(c, r));
}
});
async function fetchProfile(key: string) {

1
packages/system/package.json
View File

@ -36,6 +36,7 @@
"@snort/shared": "^1.0.7",
"@stablelib/xchacha20": "^1.0.1",
"debug": "^4.3.4",
"events": "^3.3.0",
"isomorphic-ws": "^5.0.0",
"uuid": "^9.0.0",
"ws": "^8.14.0"

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

@ -8,8 +8,7 @@ import { ConnectionStats } from "./connection-stats";
import { NostrEvent, ReqCommand, ReqFilter, TaggedNostrEvent, u256 } from "./nostr";
import { RelayInfo } from "./relay-info";
import EventKind from "./event-kind";
export type AuthHandler = (challenge: string, relay: string) => Promise<NostrEvent | undefined>;
import EventEmitter from "events";
/**
* Relay settings
@ -46,7 +45,22 @@ export interface ConnectionStateSnapshot {
address: string;
}
export class Connection extends ExternalStore<ConnectionStateSnapshot> {
interface ConnectionEvents {
change: (snapshot: ConnectionStateSnapshot) => void;
connected: (wasReconnect: boolean) => void;
event: (sub: string, e: TaggedNostrEvent) => void;
eose: (sub: string) => void;
disconnect: (code: number) => void;
auth: (challenge: string, relay: string, cb: (ev: NostrEvent) => void) => void;
notice: (msg: string) => void;
}
export declare interface Connection {
on<U extends keyof ConnectionEvents>(event: U, listener: ConnectionEvents[U]): this;
once<U extends keyof ConnectionEvents>(event: U, listener: ConnectionEvents[U]): this;
}
export class Connection extends EventEmitter {
#log: debug.Debugger;
#ephemeralCheck?: ReturnType<typeof setInterval>;
#activity: number = unixNowMs();
@ -72,16 +86,12 @@ export class Connection extends ExternalStore<ConnectionStateSnapshot> {
IsClosed: boolean;
ReconnectTimer?: ReturnType<typeof setTimeout>;
EventsCallback: Map<u256, (msg: Array<string | boolean>) => void>;
OnConnected?: (wasReconnect: boolean) => void;
OnEvent?: (sub: string, e: TaggedNostrEvent) => void;
OnEose?: (sub: string) => void;
OnDisconnect?: (code: number) => void;
Auth?: AuthHandler;
AwaitingAuth: Map<string, boolean>;
Authed = false;
Down = true;
constructor(addr: string, options: RelaySettings, auth?: AuthHandler, ephemeral: boolean = false) {
constructor(addr: string, options: RelaySettings, ephemeral: boolean = false) {
super();
this.Id = uuid();
this.Address = addr;
@ -89,7 +99,6 @@ export class Connection extends ExternalStore<ConnectionStateSnapshot> {
this.IsClosed = false;
this.EventsCallback = new Map();
this.AwaitingAuth = new Map();
this.Auth = auth;
this.#ephemeral = ephemeral;
this.#log = debug("Connection").extend(addr);
}
@ -154,7 +163,7 @@ export class Connection extends ExternalStore<ConnectionStateSnapshot> {
this.#log(`Open!`);
this.Down = false;
this.#setupEphemeral();
this.OnConnected?.(wasReconnect);
this.emit("connected", wasReconnect);
this.#sendPendingRaw();
}
@ -182,7 +191,7 @@ export class Connection extends ExternalStore<ConnectionStateSnapshot> {
this.ReconnectTimer = undefined;
}
this.OnDisconnect?.(e.code);
this.emit("disconnected", e.code);
this.#reset();
this.notifyChange();
}
@ -206,7 +215,7 @@ export class Connection extends ExternalStore<ConnectionStateSnapshot> {
break;
}
case "EVENT": {
this.OnEvent?.(msg[1] as string, {
this.emit("event", msg[1] as string, {
...(msg[2] as NostrEvent),
relays: [this.Address],
});
@ -215,7 +224,7 @@ export class Connection extends ExternalStore<ConnectionStateSnapshot> {
break;
}
case "EOSE": {
this.OnEose?.(msg[1] as string);
this.emit("eose", msg[1] as string);
break;
}
case "OK": {
@ -230,6 +239,7 @@ export class Connection extends ExternalStore<ConnectionStateSnapshot> {
break;
}
case "NOTICE": {
this.emit("notice", msg[1]);
this.#log(`NOTICE: ${msg[1]}`);
break;
}
@ -346,7 +356,7 @@ export class Connection extends ExternalStore<ConnectionStateSnapshot> {
CloseReq(id: string) {
if (this.ActiveRequests.delete(id)) {
this.#sendJson(["CLOSE", id]);
this.OnEose?.(id);
this.emit("eose", id);
this.#SendQueuedRequests();
}
this.notifyChange();
@ -435,11 +445,11 @@ export class Connection extends ExternalStore<ConnectionStateSnapshot> {
const authCleanup = () => {
this.AwaitingAuth.delete(challenge);
};
if (!this.Auth) {
throw new Error("Auth hook not registered");
}
this.AwaitingAuth.set(challenge, true);
const authEvent = await this.Auth(challenge, this.Address);
const authEvent = await new Promise<NostrEvent>((resolve, reject) =>
this.emit("auth", challenge, this.Address, resolve),
);
this.#log("Auth result: %o", authEvent);
if (!authEvent) {
authCleanup();
throw new Error("No auth event");
@ -486,4 +496,8 @@ export class Connection extends ExternalStore<ConnectionStateSnapshot> {
}, 5_000);
}
}
notifyChange() {
this.emit("change", this.takeSnapshot());
}
}

8
packages/system/src/impl/nip46.ts
View File

@ -85,10 +85,10 @@ export class Nip46Signer implements EventSigner {
}
return await new Promise<void>((resolve, reject) => {
this.#conn = new Connection(this.#relay, { read: true, write: true });
this.#conn.OnEvent = async (sub, e) => {
this.#conn.on("event", async (sub, e) => {
await this.#onReply(e);
};
this.#conn.OnConnected = async () => {
});
this.#conn.on("connected", async () => {
this.#conn!.QueueReq(
[
"REQ",
@ -110,7 +110,7 @@ export class Nip46Signer implements EventSigner {
resolve,
});
}
};
});
this.#conn.Connect();
this.#didInit = true;
});

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

@ -1,4 +1,4 @@
import { AuthHandler, RelaySettings, ConnectionStateSnapshot, OkResponse } from "./connection";
import { RelaySettings, ConnectionStateSnapshot, OkResponse } from "./connection";
import { RequestBuilder } from "./request-builder";
import { NoteStore, NoteStoreSnapshotData } from "./note-collection";
import { Query } from "./query";
@ -46,11 +46,6 @@ export interface SystemInterface {
*/
checkSigs: boolean;
/**
* Handler function for NIP-42
*/
HandleAuth?: AuthHandler;
/**
* Get a snapshot of the relay connections
*/

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

@ -1,8 +1,9 @@
import debug from "debug";
import EventEmitter from "events";
import { unwrap, sanitizeRelayUrl, ExternalStore, FeedCache, removeUndefined } from "@snort/shared";
import { unwrap, sanitizeRelayUrl, FeedCache, removeUndefined } from "@snort/shared";
import { NostrEvent, TaggedNostrEvent } from "./nostr";
import { AuthHandler, Connection, RelaySettings, ConnectionStateSnapshot, OkResponse } from "./connection";
import { Connection, RelaySettings, ConnectionStateSnapshot, OkResponse } from "./connection";
import { Query } from "./query";
import { NoteCollection, NoteStore, NoteStoreSnapshotData } from "./note-collection";
import { BuiltRawReqFilter, RequestBuilder, RequestStrategy } from "./request-builder";
@ -25,10 +26,20 @@ import { RelayCache } from "./gossip-model";
import { QueryOptimizer, DefaultQueryOptimizer } from "./query-optimizer";
import { trimFilters } from "./request-trim";
interface NostrSystemEvents {
change: (state: SystemSnapshot) => void;
auth: (challenge: string, relay: string, cb: (ev: NostrEvent) => void) => void;
}
export declare interface NostrSystem {
on<U extends keyof NostrSystemEvents>(event: U, listener: NostrSystemEvents[U]): this;
once<U extends keyof NostrSystemEvents>(event: U, listener: NostrSystemEvents[U]): this;
}
/**
* Manages nostr content retrieval system
*/
export class NostrSystem extends ExternalStore<SystemSnapshot> implements SystemInterface {
export class NostrSystem extends EventEmitter implements SystemInterface {
#log = debug("System");
/**
@ -41,11 +52,6 @@ export class NostrSystem extends ExternalStore<SystemSnapshot> implements System
*/
Queries: Map<string, Query> = new Map();
/**
* NIP-42 Auth handler
*/
#handleAuth?: AuthHandler;
/**
* Storage class for user relay lists
*/
@ -87,7 +93,6 @@ export class NostrSystem extends ExternalStore<SystemSnapshot> implements System
checkSigs: boolean;
constructor(props: {
authHandler?: AuthHandler;
relayCache?: FeedCache<UsersRelays>;
profileCache?: FeedCache<MetadataCache>;
relayMetrics?: FeedCache<RelayMetrics>;
@ -97,7 +102,6 @@ export class NostrSystem extends ExternalStore<SystemSnapshot> implements System
checkSigs?: boolean;
}) {
super();
this.#handleAuth = props.authHandler;
this.#relayCache = props.relayCache ?? new UserRelaysCache(props.db?.userRelays);
this.#profileCache = props.profileCache ?? new UserProfileCache(props.db?.users);
this.#relayMetricsCache = props.relayMetrics ?? new RelayMetricCache(props.db?.relayMetrics);
@ -110,14 +114,12 @@ export class NostrSystem extends ExternalStore<SystemSnapshot> implements System
this.#cleanup();
}
HandleAuth?: AuthHandler | undefined;
get ProfileLoader() {
return this.#profileLoader;
}
get Sockets(): ConnectionStateSnapshot[] {
return [...this.#sockets.values()].map(a => a.snapshot());
return [...this.#sockets.values()].map(a => a.takeSnapshot());
}
get RelayCache(): RelayCache {
@ -149,12 +151,12 @@ export class NostrSystem extends ExternalStore<SystemSnapshot> implements System
const addr = unwrap(sanitizeRelayUrl(address));
const existing = this.#sockets.get(addr);
if (!existing) {
const c = new Connection(addr, options, this.#handleAuth?.bind(this));
const c = new Connection(addr, options);
this.#sockets.set(addr, c);
c.OnEvent = (s, e) => this.#onEvent(s, e);
c.OnEose = s => this.#onEndOfStoredEvents(c, s);
c.OnDisconnect = code => this.#onRelayDisconnect(c, code);
c.OnConnected = r => this.#onRelayConnected(c, r);
c.on("event", (s, e) => this.#onEvent(s, e));
c.on("eose", s => this.#onEndOfStoredEvents(c, s));
c.on("disconnect", code => this.#onRelayDisconnect(c, code));
c.on("connected", r => this.#onRelayConnected(c, r));
await c.Connect();
} else {
// update settings if already connected
@ -210,12 +212,12 @@ export class NostrSystem extends ExternalStore<SystemSnapshot> implements System
try {
const addr = unwrap(sanitizeRelayUrl(address));
if (!this.#sockets.has(addr)) {
const c = new Connection(addr, { read: true, write: true }, this.#handleAuth?.bind(this), true);
const c = new Connection(addr, { read: true, write: true }, true);
this.#sockets.set(addr, c);
c.OnEvent = (s, e) => this.#onEvent(s, e);
c.OnEose = s => this.#onEndOfStoredEvents(c, s);
c.OnDisconnect = code => this.#onRelayDisconnect(c, code);
c.OnConnected = r => this.#onRelayConnected(c, r);
c.on("event", (s, e) => this.#onEvent(s, e));
c.on("eose", s => this.#onEndOfStoredEvents(c, s));
c.on("disconnect", code => this.#onRelayDisconnect(c, code));
c.on("connected", r => this.#onRelayConnected(c, r));
await c.Connect();
return c;
}
@ -404,15 +406,15 @@ export class NostrSystem extends ExternalStore<SystemSnapshot> implements System
return await existing.SendAsync(ev);
} else {
return await new Promise<OkResponse>((resolve, reject) => {
const c = new Connection(address, { write: true, read: true }, this.#handleAuth?.bind(this), true);
const c = new Connection(address, { write: true, read: true }, true);
const t = setTimeout(reject, 10_000);
c.OnConnected = async () => {
c.once("connected", async () => {
clearTimeout(t);
const rsp = await c.SendAsync(ev);
c.Close();
resolve(rsp);
};
});
c.Connect();
});
}
@ -430,6 +432,10 @@ export class NostrSystem extends ExternalStore<SystemSnapshot> implements System
};
}
notifyChange() {
this.emit("change", this.takeSnapshot());
}
#cleanup() {
let changed = false;
for (const [k, v] of this.Queries) {

1
yarn.lock
View File

@ -3243,6 +3243,7 @@ __metadata:
"@types/uuid": ^9.0.2
"@types/ws": ^8.5.5
debug: ^4.3.4
events: ^3.3.0
isomorphic-ws: ^5.0.0
jest: ^29.5.0
jest-environment-jsdom: ^29.5.0