From 0074b51abaf9fc3134df842528666e5f338cb164 Mon Sep 17 00:00:00 2001 From: Jon Staab Date: Thu, 7 Nov 2024 09:07:03 -0800 Subject: [PATCH] Re-work feed to reduce annoyance of calling loaders --- packages/app/src/feeds.ts | 108 +++++++-------- packages/feeds/README.md | 22 ++-- .../feeds/src/{loader.ts => controller.ts} | 124 ++++++++++-------- packages/feeds/src/core.ts | 4 + packages/feeds/src/index.ts | 2 +- packages/util/src/Filters.ts | 26 +++- 6 files changed, 165 insertions(+), 121 deletions(-) rename packages/feeds/src/{loader.ts => controller.ts} (62%) diff --git a/packages/app/src/feeds.ts b/packages/app/src/feeds.ts index 7a0acae..8edc464 100644 --- a/packages/app/src/feeds.ts +++ b/packages/app/src/feeds.ts @@ -1,6 +1,7 @@ import {ctx, now} from '@welshman/lib' import {createEvent, getPubkeyTagValues} from '@welshman/util' -import {FeedLoader, Scope} from '@welshman/feeds' +import {Scope} from '@welshman/feeds' +import type {RequestOpts, DVMOpts} from '@welshman/feeds' import {makeDvmRequest} from '@welshman/dvm' import {makeSecret, Nip01Signer} from '@welshman/signer' import {pubkey, signer} from './session' @@ -8,61 +9,62 @@ import {getFilterSelections} from './router' import {wotGraph, maxWot, getFollows, getNetwork, getFollowers} from './wot' import {load} from './core' -export const feedLoader = new FeedLoader({ - request: async ({filters = [{}], relays = [], onEvent}) => { - if (relays.length > 0) { - await load({onEvent, filters, relays}) - } else { - await Promise.all( - getFilterSelections(filters) - .map(opts => load({onEvent, ...opts})) - ) - } - }, - requestDVM: async ({kind, onEvent, ...request}) => { - const tags = [...request.tags || [], ["expiration", String(now() + 5)]] - const $signer = signer.get() || new Nip01Signer(makeSecret()) - const event = await $signer.sign(createEvent(kind, {tags})) - const relays = - request.relays - ? ctx.app.router.FromRelays(request.relays).getUrls() - : ctx.app.router.FromPubkeys(getPubkeyTagValues(tags)).getUrls() +export const request = async ({filters = [{}], relays = [], onEvent}: RequestOpts) => { + if (relays.length > 0) { + await load({onEvent, filters, relays}) + } else { + await Promise.all( + getFilterSelections(filters) + .map(opts => load({onEvent, ...opts})) + ) + } +} - const req = makeDvmRequest({event, relays}) +export const requestDVM = async ({kind, onEvent, ...request}: DVMOpts) => { + const tags = [...request.tags || [], ["expiration", String(now() + 5)]] + const $signer = signer.get() || new Nip01Signer(makeSecret()) + const event = await $signer.sign(createEvent(kind, {tags})) + const relays = + request.relays + ? ctx.app.router.FromRelays(request.relays).getUrls() + : ctx.app.router.FromPubkeys(getPubkeyTagValues(tags)).getUrls() - await new Promise(resolve => { - req.emitter.on("result", (url, event) => { - onEvent(event) - resolve() - }) + const req = makeDvmRequest({event, relays}) + + await new Promise(resolve => { + req.emitter.on("result", (url, event) => { + onEvent(event) + resolve() }) - }, - getPubkeysForScope: (scope: string) => { - const $pubkey = pubkey.get() + }) +} - if (!$pubkey) { - return [] +export const getPubkeysForScope = (scope: string) => { + const $pubkey = pubkey.get() + + if (!$pubkey) { + return [] + } + + switch (scope) { + case Scope.Self: return [$pubkey] + case Scope.Follows: return getFollows($pubkey) + case Scope.Network: return getNetwork($pubkey) + case Scope.Followers: return getFollowers($pubkey) + default: return [] + } +} + +export const getPubkeysForWOTRange = (min: number, max: number) => { + const pubkeys = [] + const thresholdMin = maxWot.get() * min + const thresholdMax = maxWot.get() * max + + for (const [tpk, score] of wotGraph.get().entries()) { + if (score >= thresholdMin && score <= thresholdMax) { + pubkeys.push(tpk) } + } - switch (scope) { - case Scope.Self: return [$pubkey] - case Scope.Follows: return getFollows($pubkey) - case Scope.Network: return getNetwork($pubkey) - case Scope.Followers: return getFollowers($pubkey) - default: return [] - } - }, - getPubkeysForWOTRange: (min, max) => { - const pubkeys = [] - const thresholdMin = maxWot.get() * min - const thresholdMax = maxWot.get() * max - - for (const [tpk, score] of wotGraph.get().entries()) { - if (score >= thresholdMin && score <= thresholdMax) { - pubkeys.push(tpk) - } - } - - return pubkeys - }, -}) + return pubkeys +} diff --git a/packages/feeds/README.md b/packages/feeds/README.md index 3e26170..272d313 100644 --- a/packages/feeds/README.md +++ b/packages/feeds/README.md @@ -5,14 +5,6 @@ A custom feed compiler and loader for nostr. Read the spec on [wikifreedia](http # Example ```javascript -// Configure the feed loader so it can access your app's context and make requests -const loader = new FeedLoader({ - request, - requestDvm, - getPubkeysForScope, - getPubkeysForWotRange, -}) - // Define a feed using set operations const feed = intersectionFeed( unionFeed( @@ -26,9 +18,17 @@ const feed = intersectionFeed( scopeFeed("global"), ) -// Load notes using the feed -loader.compiler.getLoader(feed, { +// Create a controller +const controller = new FeedController({ + feed, + request, + requestDvm, + getPubkeysForScope, + getPubkeysForWotRange, onEvent: event => console.log("Event", event), - onExhausted: () => console.log("Exhausted"), + onExhausted: () => console.log("Exhausted"), }) + +// Load notes using the feed +controller.load(10) ``` diff --git a/packages/feeds/src/loader.ts b/packages/feeds/src/controller.ts similarity index 62% rename from packages/feeds/src/loader.ts rename to packages/feeds/src/controller.ts index d591479..76d35c1 100644 --- a/packages/feeds/src/loader.ts +++ b/packages/feeds/src/controller.ts @@ -1,53 +1,56 @@ -import {inc, omitVals, max, min, now} from '@welshman/lib' +import {inc, memoize, omitVals, max, min, now} from '@welshman/lib' import type {TrustedEvent, Filter} from '@welshman/util' import {EPOCH, trimFilters, guessFilterDelta} from '@welshman/util' import type {Feed, RequestItem, FeedOptions} from './core' import {FeedType} from './core' import {FeedCompiler} from './compiler' -export type LoadOpts = { - onEvent?: (event: TrustedEvent) => void - onExhausted?: () => void - useWindowing?: boolean -} - -export type Loader = (limit: number) => Promise - -export class FeedLoader { +export class FeedController { compiler: FeedCompiler constructor(readonly options: FeedOptions) { this.compiler = new FeedCompiler(options) } - async getLoader([type, ...feed]: Feed, loadOpts: LoadOpts = {}) { - if (this.compiler.canCompile([type, ...feed] as Feed)) { - return this.getRequestsLoader(await this.compiler.compile([type, ...feed] as Feed), loadOpts) + getFilters = memoize(async () => { + return this.compiler.canCompile(this.options.feed) + ? this.compiler.compile(this.options.feed) + : undefined + }) + + getLoader = memoize(async () => { + const [type, ...feed] = this.options.feed + const filters = await this.getFilters() + + if (filters) { + return this._getRequestsLoader(filters) } switch(type) { case FeedType.Difference: - return this._getDifferenceLoader(feed as Feed[], loadOpts) + return this._getDifferenceLoader(feed as Feed[]) case FeedType.Intersection: - return this._getIntersectionLoader(feed as Feed[], loadOpts) + return this._getIntersectionLoader(feed as Feed[]) case FeedType.Union: - return this._getUnionLoader(feed as Feed[], loadOpts) + return this._getUnionLoader(feed as Feed[]) default: throw new Error(`Unable to convert feed of type ${type} to loader`) } - } + }) - async getRequestsLoader(requests: RequestItem[], loadOpts: LoadOpts) { + load = async (limit: number) => (await this.getLoader())(limit) + + async _getRequestsLoader(requests: RequestItem[], overrides: Partial = {}) { + const {onEvent, onExhausted} = {...this.options, ...overrides} const seen = new Set() const exhausted = new Set() const loaders = await Promise.all( requests.map( request => this._getRequestLoader(request, { - ...loadOpts, onExhausted: () => exhausted.add(request), onEvent: e => { if (!seen.has(e.id)) { - loadOpts.onEvent?.(e) + onEvent(e) seen.add(e.id) } }, @@ -59,12 +62,14 @@ export class FeedLoader { await Promise.all(loaders.map(loader => loader(limit))) if (exhausted.size === requests.length) { - loadOpts.onExhausted?.() + onExhausted() } } } - async _getRequestLoader({relays, filters}: RequestItem, {useWindowing = true, onEvent, onExhausted}: LoadOpts) { + async _getRequestLoader({relays, filters}: RequestItem, overrides: Partial = {}) { + const {useWindowing, onEvent, onExhausted, request} = {...this.options, ...overrides} + // Make sure we have some kind of filter to send if we've been given an empty one, as happens with relay feeds if (!filters || filters.length === 0) { filters = [{}] @@ -93,24 +98,24 @@ export class FeedLoader { .map((filter: Filter) => ({...filter, until, limit, since})) if (requestFilters.length === 0) { - return onExhausted?.() + return onExhausted() } let count = 0 - await this.options.request(omitVals([undefined], { + await request(omitVals([undefined], { relays, filters: trimFilters(requestFilters), onEvent: (event: TrustedEvent) => { count += 1 until = Math.min(until, event.created_at - 1) - onEvent?.(event) + onEvent(event) }, })) if (useWindowing) { if (since === minSince) { - onExhausted?.() + onExhausted() } // Relays can't be relied upon to return events in descending order, do exponential @@ -122,20 +127,23 @@ export class FeedLoader { since = Math.max(minSince, until - delta) } else if (count === 0) { - onExhausted?.() + onExhausted() } } } - async _getDifferenceLoader(feeds: Feed[], {onEvent, onExhausted}: LoadOpts) { + async _getDifferenceLoader(feeds: Feed[], overrides: Partial = {}) { + const {onEvent, onExhausted, ...options} = {...this.options, ...overrides} const exhausted = new Set() const skip = new Set() const events: TrustedEvent[] = [] const seen = new Set() - const loaders = await Promise.all( - feeds.map((feed: Feed, i: number) => - this.getLoader(feed, { + const controllers = await Promise.all( + feeds.map((thisFeed: Feed, i: number) => + new FeedController({ + ...options, + feed: thisFeed, onExhausted: () => exhausted.add(i), onEvent: (event: TrustedEvent) => { if (i === 0) { @@ -150,37 +158,40 @@ export class FeedLoader { return async (limit: number) => { await Promise.all( - loaders.map(async (loader: Loader, i: number) => { + controllers.map(async (controller: FeedController, i: number) => { if (exhausted.has(i)) { return } - await loader(limit) + await controller.load(limit) }) ) for (const event of events.splice(0)) { if (!skip.has(event.id) && !seen.has(event.id)) { - onEvent?.(event) + onEvent(event) seen.add(event.id) } } - if (exhausted.size === loaders.length) { - onExhausted?.() + if (exhausted.size === controllers.length) { + onExhausted() } } } - async _getIntersectionLoader(feeds: Feed[], {onEvent, onExhausted}: LoadOpts) { + async _getIntersectionLoader(feeds: Feed[], overrides: Partial = {}) { + const {onEvent, onExhausted, ...options} = {...this.options, ...overrides} const exhausted = new Set() const counts = new Map() const events: TrustedEvent[] = [] const seen = new Set() - const loaders = await Promise.all( - feeds.map((feed: Feed, i: number) => - this.getLoader(feed, { + const controllers = await Promise.all( + feeds.map((thisFeed: Feed, i: number) => + new FeedController({ + ...options, + feed: thisFeed, onExhausted: () => exhausted.add(i), onEvent: (event: TrustedEvent) => { events.push(event) @@ -192,39 +203,42 @@ export class FeedLoader { return async (limit: number) => { await Promise.all( - loaders.map(async (loader: Loader, i: number) => { + controllers.map(async (controller: FeedController, i: number) => { if (exhausted.has(i)) { return } - await loader(limit) + await controller.load(limit) }) ) for (const event of events.splice(0)) { - if (counts.get(event.id) === loaders.length && !seen.has(event.id)) { - onEvent?.(event) + if (counts.get(event.id) === controllers.length && !seen.has(event.id)) { + onEvent(event) seen.add(event.id) } } - if (exhausted.size === loaders.length) { - onExhausted?.() + if (exhausted.size === controllers.length) { + onExhausted() } } } - async _getUnionLoader(feeds: Feed[], {onEvent, onExhausted}: LoadOpts) { + async _getUnionLoader(feeds: Feed[], overrides: Partial = {}) { + const {onEvent, onExhausted, ...options} = {...this.options, ...overrides} const exhausted = new Set() const seen = new Set() - const loaders = await Promise.all( - feeds.map((feed: Feed, i: number) => - this.getLoader(feed, { + const controllers = await Promise.all( + feeds.map((thisFeed: Feed, i: number) => + new FeedController({ + ...options, + feed: thisFeed, onExhausted: () => exhausted.add(i), onEvent: (event: TrustedEvent) => { if (!seen.has(event.id)) { - onEvent?.(event) + onEvent(event) seen.add(event.id) } }, @@ -234,17 +248,17 @@ export class FeedLoader { return async (limit: number) => { await Promise.all( - loaders.map(async (loader: Loader, i: number) => { + controllers.map(async (controller: FeedController, i: number) => { if (exhausted.has(i)) { return } - await loader(limit) + await controller.load(limit) }) ) - if (exhausted.size === loaders.length) { - onExhausted?.() + if (exhausted.size === controllers.length) { + onExhausted() } } } diff --git a/packages/feeds/src/core.ts b/packages/feeds/src/core.ts index 6710d22..3657c54 100644 --- a/packages/feeds/src/core.ts +++ b/packages/feeds/src/core.ts @@ -124,8 +124,12 @@ export type DVMOpts = DVMRequest & { } export type FeedOptions = { + feed: Feed request: (opts: RequestOpts) => Promise requestDVM: (opts: DVMOpts) => Promise getPubkeysForScope: (scope: Scope) => string[] getPubkeysForWOTRange: (minWOT: number, maxWOT: number) => string[] + onEvent: (event: TrustedEvent) => void + onExhausted: () => void + useWindowing?: boolean } diff --git a/packages/feeds/src/index.ts b/packages/feeds/src/index.ts index afb416d..f25e136 100644 --- a/packages/feeds/src/index.ts +++ b/packages/feeds/src/index.ts @@ -1,4 +1,4 @@ export * from './core' export * from './compiler' -export * from './loader' +export * from './controller' export * from './utils' diff --git a/packages/util/src/Filters.ts b/packages/util/src/Filters.ts index eca0e1c..edc35c1 100644 --- a/packages/util/src/Filters.ts +++ b/packages/util/src/Filters.ts @@ -1,11 +1,13 @@ import {matchFilter as nostrToolsMatchFilter} from 'nostr-tools' -import {uniqBy, prop, mapVals, shuffle, avg, hash, groupBy, randomId, uniq} from '@welshman/lib' +import {without, uniqBy, prop, mapVals, shuffle, avg, hash, groupBy, randomId, uniq} from '@welshman/lib' import type {HashedEvent, TrustedEvent, SignedEvent} from './Events' import {isReplaceableKind} from './Kinds' import {Address, getAddress} from './Address' export const EPOCH = 1609459200 +export const neverFilter = {ids: []} + export type Filter = { ids?: string[] kinds?: number[] @@ -184,6 +186,28 @@ export const getReplyFilters = (events: TrustedEvent[], filter: Filter = {}) => return filters } + +export const addRepostFilters = (filters: Filter[]) => + filters.flatMap(original => { + const filterChunk = [original] + + if (!original.kinds) { + filterChunk.push({...original, kinds: [6, 16]}) + } else { + if (original.kinds.includes(1)) { + filterChunk.push({...original, kinds: [6]}) + } + + const otherKinds = without([1], original.kinds) + + if (otherKinds.length > 0) { + filterChunk.push({...original, kinds: [16], "#k": otherKinds.map(String)}) + } + } + + return filterChunk + }) + export const getFilterGenerality = (filter: Filter) => { if (filter.ids || filter["#e"] || filter["#a"]) { return 0