Delete accidentally added file
This commit is contained in:
parent
bf91d0393e
commit
375fe09cdf
|
@ -1,307 +0,0 @@
|
||||||
[1mdiff --git a/src/client.ts b/src/client.ts[m
|
|
||||||
[1mindex 970a077..3cf2e8a 100644[m
|
|
||||||
[1m--- a/src/client.ts[m
|
|
||||||
[1m+++ b/src/client.ts[m
|
|
||||||
[36m@@ -14,7 +14,7 @@[m [mfunction getFilters<K extends number>(filters: Filter<K>[], opts: GetFiltersOpts[m
|
|
||||||
[m
|
|
||||||
const unsub = pool.subscribe([m
|
|
||||||
filters,[m
|
|
||||||
[31m- activeRelays,[m
|
|
||||||
[32m+[m[32m opts.relays ?? activeRelays,[m
|
|
||||||
(event: Event | null) => {[m
|
|
||||||
if (event && matchFilters(filters, event)) {[m
|
|
||||||
pipeline.handleEvent(event).catch(() => {});[m
|
|
||||||
[1mdiff --git a/src/common.ts b/src/common.ts[m
|
|
||||||
[1mnew file mode 100644[m
|
|
||||||
[1mindex 0000000..0424b52[m
|
|
||||||
[1m--- /dev/null[m
|
|
||||||
[1m+++ b/src/common.ts[m
|
|
||||||
[36m@@ -0,0 +1,9 @@[m
|
|
||||||
[32m+[m[32mimport { Reqmeister } from '@/reqmeister.ts';[m
|
|
||||||
[32m+[m[32mimport { Time } from '@/utils/time.ts';[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32mconst reqmeister = new Reqmeister({[m
|
|
||||||
[32m+[m[32m delay: Time.seconds(1),[m
|
|
||||||
[32m+[m[32m signal: AbortSignal.timeout(Time.seconds(1)),[m
|
|
||||||
[32m+[m[32m});[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32mexport { reqmeister };[m
|
|
||||||
[1mdiff --git a/src/deps.ts b/src/deps.ts[m
|
|
||||||
[1mindex b9db9e2..4a9314b 100644[m
|
|
||||||
[1m--- a/src/deps.ts[m
|
|
||||||
[1m+++ b/src/deps.ts[m
|
|
||||||
[36m@@ -81,5 +81,7 @@[m [mexport { Machina } from 'https://gitlab.com/soapbox-pub/nostr-machina/-/raw/08a1[m
|
|
||||||
export * as Sentry from 'https://deno.land/x/sentry@7.78.0/index.js';[m
|
|
||||||
export { sentry as sentryMiddleware } from 'npm:@hono/sentry@^1.0.0';[m
|
|
||||||
export * as Comlink from 'npm:comlink@^4.4.1';[m
|
|
||||||
[32m+[m[32mexport { EventEmitter } from 'npm:tseep@^1.1.3';[m
|
|
||||||
[32m+[m[32mexport { default as stringifyStable } from 'npm:fast-stable-stringify@^1.0.0';[m
|
|
||||||
[m
|
|
||||||
export type * as TypeFest from 'npm:type-fest@^4.3.0';[m
|
|
||||||
[1mdiff --git a/src/filter.ts b/src/filter.ts[m
|
|
||||||
[1mindex fb43251..38fcff7 100644[m
|
|
||||||
[1m--- a/src/filter.ts[m
|
|
||||||
[1m+++ b/src/filter.ts[m
|
|
||||||
[36m@@ -1,5 +1,5 @@[m
|
|
||||||
import { Conf } from '@/config.ts';[m
|
|
||||||
[31m-import { type Event, type Filter, matchFilters } from '@/deps.ts';[m
|
|
||||||
[32m+[m[32mimport { type Event, type Filter, matchFilters, stringifyStable } from '@/deps.ts';[m
|
|
||||||
[m
|
|
||||||
import type { EventData } from '@/types.ts';[m
|
|
||||||
[m
|
|
||||||
[36m@@ -14,12 +14,17 @@[m [minterface DittoFilter<K extends number = number> extends Filter<K> {[m
|
|
||||||
relations?: Relation[];[m
|
|
||||||
}[m
|
|
||||||
[m
|
|
||||||
[32m+[m[32m/** Filter to get one specific event. */[m
|
|
||||||
[32m+[m[32mtype MicroFilter = { ids: [Event['id']] } | { kinds: [0]; authors: [Event['pubkey']] };[m
|
|
||||||
[32m+[m
|
|
||||||
/** Additional options to apply to the whole subscription. */[m
|
|
||||||
interface GetFiltersOpts {[m
|
|
||||||
/** Signal to abort the request. */[m
|
|
||||||
signal?: AbortSignal;[m
|
|
||||||
/** Event limit for the whole subscription. */[m
|
|
||||||
limit?: number;[m
|
|
||||||
[32m+[m[32m /** Relays to use, if applicable. */[m
|
|
||||||
[32m+[m[32m relays?: WebSocket['url'][];[m
|
|
||||||
}[m
|
|
||||||
[m
|
|
||||||
function matchDittoFilter(filter: DittoFilter, event: Event, data: EventData): boolean {[m
|
|
||||||
[36m@@ -44,4 +49,33 @@[m [mfunction matchDittoFilters(filters: DittoFilter[], event: Event, data: EventData[m
|
|
||||||
return false;[m
|
|
||||||
}[m
|
|
||||||
[m
|
|
||||||
[31m-export { type DittoFilter, type GetFiltersOpts, matchDittoFilters, type Relation };[m
|
|
||||||
[32m+[m[32m/** Get deterministic ID for a microfilter. */[m
|
|
||||||
[32m+[m[32mfunction getFilterId(filter: MicroFilter): string {[m
|
|
||||||
[32m+[m[32m if ('ids' in filter) {[m
|
|
||||||
[32m+[m[32m return stringifyStable({ ids: [filter.ids] });[m
|
|
||||||
[32m+[m[32m } else {[m
|
|
||||||
[32m+[m[32m return stringifyStable({[m
|
|
||||||
[32m+[m[32m kinds: [filter.kinds[0]],[m
|
|
||||||
[32m+[m[32m authors: [filter.authors[0]],[m
|
|
||||||
[32m+[m[32m });[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m[32m}[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m/** Get a microfilter from a Nostr event. */[m
|
|
||||||
[32m+[m[32mfunction eventToMicroFilter(event: Event): MicroFilter {[m
|
|
||||||
[32m+[m[32m if (event.kind === 0) {[m
|
|
||||||
[32m+[m[32m return { kinds: [0], authors: [event.pubkey] };[m
|
|
||||||
[32m+[m[32m } else {[m
|
|
||||||
[32m+[m[32m return { ids: [event.id] };[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m[32m}[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32mexport {[m
|
|
||||||
[32m+[m[32m type DittoFilter,[m
|
|
||||||
[32m+[m[32m eventToMicroFilter,[m
|
|
||||||
[32m+[m[32m getFilterId,[m
|
|
||||||
[32m+[m[32m type GetFiltersOpts,[m
|
|
||||||
[32m+[m[32m matchDittoFilters,[m
|
|
||||||
[32m+[m[32m type MicroFilter,[m
|
|
||||||
[32m+[m[32m type Relation,[m
|
|
||||||
[32m+[m[32m};[m
|
|
||||||
[1mdiff --git a/src/pipeline.ts b/src/pipeline.ts[m
|
|
||||||
[1mindex adf8a84..923bf4e 100644[m
|
|
||||||
[1m--- a/src/pipeline.ts[m
|
|
||||||
[1m+++ b/src/pipeline.ts[m
|
|
||||||
[36m@@ -1,3 +1,4 @@[m
|
|
||||||
[32m+[m[32mimport { reqmeister } from '@/common.ts';[m
|
|
||||||
import { Conf } from '@/config.ts';[m
|
|
||||||
import * as eventsDB from '@/db/events.ts';[m
|
|
||||||
import { addRelays } from '@/db/relays.ts';[m
|
|
||||||
[36m@@ -23,15 +24,17 @@[m [mimport type { EventData } from '@/types.ts';[m
|
|
||||||
*/[m
|
|
||||||
async function handleEvent(event: Event): Promise<void> {[m
|
|
||||||
if (!(await verifySignatureWorker(event))) return;[m
|
|
||||||
[32m+[m[32m const wanted = reqmeister.isWanted(event);[m
|
|
||||||
if (encounterEvent(event)) return;[m
|
|
||||||
console.info(`pipeline: Event<${event.kind}> ${event.id}`);[m
|
|
||||||
const data = await getEventData(event);[m
|
|
||||||
[m
|
|
||||||
await Promise.all([[m
|
|
||||||
[31m- storeEvent(event, data),[m
|
|
||||||
[32m+[m[32m storeEvent(event, data, { force: wanted }),[m
|
|
||||||
processDeletions(event),[m
|
|
||||||
trackRelays(event),[m
|
|
||||||
trackHashtags(event),[m
|
|
||||||
[32m+[m[32m fetchRelatedEvents(event, data),[m
|
|
||||||
processMedia(event, data),[m
|
|
||||||
streamOut(event, data),[m
|
|
||||||
broadcast(event, data),[m
|
|
||||||
[36m@@ -39,13 +42,14 @@[m [masync function handleEvent(event: Event): Promise<void> {[m
|
|
||||||
}[m
|
|
||||||
[m
|
|
||||||
/** Tracks encountered events to skip duplicates, improving idempotency and performance. */[m
|
|
||||||
[31m-const encounters = new LRUCache<string, boolean>({ max: 1000 });[m
|
|
||||||
[32m+[m[32mconst encounters = new LRUCache<Event['id'], true>({ max: 1000 });[m
|
|
||||||
[m
|
|
||||||
/** Encounter the event, and return whether it has already been encountered. */[m
|
|
||||||
[31m-function encounterEvent(event: Event) {[m
|
|
||||||
[32m+[m[32mfunction encounterEvent(event: Event): boolean {[m
|
|
||||||
const result = encounters.get(event.id);[m
|
|
||||||
encounters.set(event.id, true);[m
|
|
||||||
[31m- return result;[m
|
|
||||||
[32m+[m[32m reqmeister.encounter(event);[m
|
|
||||||
[32m+[m[32m return !!result;[m
|
|
||||||
}[m
|
|
||||||
[m
|
|
||||||
/** Preload data that will be useful to several tasks. */[m
|
|
||||||
[36m@@ -57,11 +61,16 @@[m [masync function getEventData({ pubkey }: Event): Promise<EventData> {[m
|
|
||||||
/** Check if the pubkey is the `DITTO_NSEC` pubkey. */[m
|
|
||||||
const isAdminEvent = ({ pubkey }: Event): boolean => pubkey === Conf.pubkey;[m
|
|
||||||
[m
|
|
||||||
[32m+[m[32minterface StoreEventOpts {[m
|
|
||||||
[32m+[m[32m force?: boolean;[m
|
|
||||||
[32m+[m[32m}[m
|
|
||||||
[32m+[m
|
|
||||||
/** Maybe store the event, if eligible. */[m
|
|
||||||
[31m-async function storeEvent(event: Event, data: EventData): Promise<void> {[m
|
|
||||||
[32m+[m[32masync function storeEvent(event: Event, data: EventData, opts: StoreEventOpts = {}): Promise<void> {[m
|
|
||||||
if (isEphemeralKind(event.kind)) return;[m
|
|
||||||
[32m+[m[32m const { force = false } = opts;[m
|
|
||||||
[m
|
|
||||||
[31m- if (data.user || isAdminEvent(event) || await isLocallyFollowed(event.pubkey)) {[m
|
|
||||||
[32m+[m[32m if (force || data.user || isAdminEvent(event) || await isLocallyFollowed(event.pubkey)) {[m
|
|
||||||
const [deletion] = await mixer.getFilters([m
|
|
||||||
[{ kinds: [5], authors: [event.pubkey], '#e': [event.id], limit: 1 }],[m
|
|
||||||
{ limit: 1, signal: AbortSignal.timeout(Time.seconds(1)) },[m
|
|
||||||
[36m@@ -129,6 +138,18 @@[m [mfunction trackRelays(event: Event) {[m
|
|
||||||
return addRelays([...relays]);[m
|
|
||||||
}[m
|
|
||||||
[m
|
|
||||||
[32m+[m[32m/** Queue related events to fetch. */[m
|
|
||||||
[32m+[m[32mfunction fetchRelatedEvents(event: Event, data: EventData) {[m
|
|
||||||
[32m+[m[32m if (!data.user) {[m
|
|
||||||
[32m+[m[32m reqmeister.req({ kinds: [0], authors: [event.pubkey] }).catch(() => {});[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m[32m for (const [name, id, relay] of event.tags) {[m
|
|
||||||
[32m+[m[32m if (name === 'e' && !encounters.has(id)) {[m
|
|
||||||
[32m+[m[32m reqmeister.req({ ids: [id] }, [relay]).catch(() => {});[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m[32m}[m
|
|
||||||
[32m+[m
|
|
||||||
/** Delete unattached media entries that are attached to the event. */[m
|
|
||||||
function processMedia({ tags, pubkey }: Event, { user }: EventData) {[m
|
|
||||||
if (user) {[m
|
|
||||||
[1mdiff --git a/src/queries.ts b/src/queries.ts[m
|
|
||||||
[1mindex fc7365a..1ecff7b 100644[m
|
|
||||||
[1m--- a/src/queries.ts[m
|
|
||||||
[1m+++ b/src/queries.ts[m
|
|
||||||
[36m@@ -2,6 +2,7 @@[m [mimport * as eventsDB from '@/db/events.ts';[m
|
|
||||||
import { type Event, findReplyTag } from '@/deps.ts';[m
|
|
||||||
import { type DittoFilter, type Relation } from '@/filter.ts';[m
|
|
||||||
import * as mixer from '@/mixer.ts';[m
|
|
||||||
[32m+[m[32mimport { reqmeister } from '@/common.ts';[m
|
|
||||||
[m
|
|
||||||
interface GetEventOpts<K extends number> {[m
|
|
||||||
/** Signal to abort the request. */[m
|
|
||||||
[36m@@ -30,10 +31,10 @@[m [mconst getEvent = async <K extends number = number>([m
|
|
||||||
const getAuthor = async (pubkey: string, opts: GetEventOpts<0> = {}): Promise<Event<0> | undefined> => {[m
|
|
||||||
const { relations, signal = AbortSignal.timeout(1000) } = opts;[m
|
|
||||||
[m
|
|
||||||
[31m- const [event] = await mixer.getFilters([m
|
|
||||||
[32m+[m[32m const event = await eventsDB.getFilters([m
|
|
||||||
[{ authors: [pubkey], relations, kinds: [0], limit: 1 }],[m
|
|
||||||
{ limit: 1, signal },[m
|
|
||||||
[31m- );[m
|
|
||||||
[32m+[m[32m ).then(([event]) => event) || await reqmeister.req({ kinds: [0], authors: [pubkey] }).catch(() => {});[m
|
|
||||||
[m
|
|
||||||
return event;[m
|
|
||||||
};[m
|
|
||||||
[1mdiff --git a/src/reqmeister.ts b/src/reqmeister.ts[m
|
|
||||||
[1mnew file mode 100644[m
|
|
||||||
[1mindex 0000000..960151f[m
|
|
||||||
[1m--- /dev/null[m
|
|
||||||
[1m+++ b/src/reqmeister.ts[m
|
|
||||||
[36m@@ -0,0 +1,88 @@[m
|
|
||||||
[32m+[m[32mimport * as client from '@/client.ts';[m
|
|
||||||
[32m+[m[32mimport { type Event, EventEmitter, type Filter } from '@/deps.ts';[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32mimport { eventToMicroFilter, getFilterId, type MicroFilter } from '@/filter.ts';[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32minterface ReqmeisterOpts {[m
|
|
||||||
[32m+[m[32m delay?: number;[m
|
|
||||||
[32m+[m[32m signal?: AbortSignal;[m
|
|
||||||
[32m+[m[32m}[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32mtype ReqmeisterQueueItem = [string, MicroFilter, WebSocket['url'][]];[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32mclass Reqmeister extends EventEmitter<{ [filterId: string]: (event: Event) => any }> {[m
|
|
||||||
[32m+[m[32m #opts: ReqmeisterOpts;[m
|
|
||||||
[32m+[m[32m #queue: ReqmeisterQueueItem[] = [];[m
|
|
||||||
[32m+[m[32m #promise!: Promise<void>;[m
|
|
||||||
[32m+[m[32m #resolve!: () => void;[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m constructor(opts: ReqmeisterOpts = {}) {[m
|
|
||||||
[32m+[m[32m super();[m
|
|
||||||
[32m+[m[32m this.#opts = opts;[m
|
|
||||||
[32m+[m[32m this.#cycle();[m
|
|
||||||
[32m+[m[32m this.#perform();[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m #cycle() {[m
|
|
||||||
[32m+[m[32m this.#resolve?.();[m
|
|
||||||
[32m+[m[32m this.#promise = new Promise((resolve) => {[m
|
|
||||||
[32m+[m[32m this.#resolve = resolve;[m
|
|
||||||
[32m+[m[32m });[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m async #perform() {[m
|
|
||||||
[32m+[m[32m const { delay } = this.#opts;[m
|
|
||||||
[32m+[m[32m await new Promise((resolve) => setTimeout(resolve, delay));[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m const queue = this.#queue;[m
|
|
||||||
[32m+[m[32m this.#queue = [];[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m const wantedEvents = new Set<Event['id']>();[m
|
|
||||||
[32m+[m[32m const wantedAuthors = new Set<Event['pubkey']>();[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m // TODO: batch by relays.[m
|
|
||||||
[32m+[m[32m for (const [_filterId, filter, _relays] of queue) {[m
|
|
||||||
[32m+[m[32m if ('ids' in filter) {[m
|
|
||||||
[32m+[m[32m filter.ids.forEach((id) => wantedEvents.add(id));[m
|
|
||||||
[32m+[m[32m } else {[m
|
|
||||||
[32m+[m[32m wantedAuthors.add(filter.authors[0]);[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m const filters: Filter[] = [];[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m if (wantedEvents.size) filters.push({ ids: [...wantedEvents] });[m
|
|
||||||
[32m+[m[32m if (wantedAuthors.size) filters.push({ kinds: [0], authors: [...wantedAuthors] });[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m const events = await client.getFilters(filters, { signal: this.#opts.signal });[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m for (const event of events) {[m
|
|
||||||
[32m+[m[32m this.encounter(event);[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m this.#cycle();[m
|
|
||||||
[32m+[m[32m this.#perform();[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m req(filter: MicroFilter, relays: WebSocket['url'][] = []): Promise<Event> {[m
|
|
||||||
[32m+[m[32m const filterId = getFilterId(filter);[m
|
|
||||||
[32m+[m[32m this.#queue.push([filterId, filter, relays]);[m
|
|
||||||
[32m+[m[32m return new Promise<Event>((resolve, reject) => {[m
|
|
||||||
[32m+[m[32m this.once(filterId, resolve);[m
|
|
||||||
[32m+[m[32m this.#promise.finally(reject);[m
|
|
||||||
[32m+[m[32m });[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m encounter(event: Event): void {[m
|
|
||||||
[32m+[m[32m const filterId = getFilterId(eventToMicroFilter(event));[m
|
|
||||||
[32m+[m[32m this.#queue = this.#queue.filter(([id]) => id !== filterId);[m
|
|
||||||
[32m+[m[32m this.emit(filterId, event);[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32m isWanted(event: Event): boolean {[m
|
|
||||||
[32m+[m[32m const filterId = getFilterId(eventToMicroFilter(event));[m
|
|
||||||
[32m+[m[32m return this.#queue.some(([id]) => id === filterId);[m
|
|
||||||
[32m+[m[32m }[m
|
|
||||||
[32m+[m[32m}[m
|
|
||||||
[32m+[m
|
|
||||||
[32m+[m[32mexport { Reqmeister };[m
|
|
Loading…
Reference in New Issue