Re-work feeds
This commit is contained in:
Jon Staab
+189
-141
@@ -1,7 +1,7 @@
|
||||
import {uniq, tryCatch, now, isNil} from '@welshman/lib'
|
||||
import {uniq, flatten, pushToMapKey, intersection, ensureNumber, tryCatch, now} from '@welshman/lib'
|
||||
import type {Rumor, Filter} from '@welshman/util'
|
||||
import {Tags, intersectFilters, BOGUS_RELAY_URL, getIdFilters, unionFilters} from '@welshman/util'
|
||||
import type {RequestItem, DVMItem, Scope, Feed, DynamicFilter, FeedOptions} from './core'
|
||||
import {Tags, intersectFilters, getAddress, getIdFilters, unionFilters} from '@welshman/util'
|
||||
import type {RequestItem, TagFilterMapping, ListItem, DVMItem, Scope, Feed, FeedOptions} from './core'
|
||||
import {FeedType, getSubFeeds} from './core'
|
||||
|
||||
export class FeedCompiler<E extends Rumor> {
|
||||
@@ -17,184 +17,232 @@ export class FeedCompiler<E extends Rumor> {
|
||||
|
||||
canCompile([type, ...feed]: Feed): boolean {
|
||||
switch(type) {
|
||||
case FeedType.Relay:
|
||||
case FeedType.Union:
|
||||
case FeedType.Intersection:
|
||||
return getSubFeeds([type, ...feed] as Feed).every(this.canCompile)
|
||||
case FeedType.Filter:
|
||||
case FeedType.List:
|
||||
case FeedType.Address:
|
||||
case FeedType.Author:
|
||||
case FeedType.DVM:
|
||||
case FeedType.ID:
|
||||
case FeedType.Kind:
|
||||
case FeedType.List:
|
||||
case FeedType.Relay:
|
||||
case FeedType.Scope:
|
||||
case FeedType.Since:
|
||||
case FeedType.SinceAgo:
|
||||
case FeedType.ATag:
|
||||
case FeedType.ETag:
|
||||
case FeedType.PTag:
|
||||
case FeedType.TTag:
|
||||
case FeedType.Until:
|
||||
case FeedType.UntilAgo:
|
||||
case FeedType.WOT:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
async compile([type, ...feed]: Feed) {
|
||||
async compile([type, ...feed]: Feed): Promise<RequestItem[]> {
|
||||
switch(type) {
|
||||
case FeedType.Union:
|
||||
return await this._compileUnion(feed as Feed[])
|
||||
case FeedType.Intersection:
|
||||
return await this._compileIntersection(feed as Feed[])
|
||||
case FeedType.Relay:
|
||||
/* eslint no-case-declarations: 0 */
|
||||
const {relays, filters} = await this._compileUnion(feed.slice(1) as Feed[])
|
||||
|
||||
return {relays: relays.concat(feed[0] as string[]), filters}
|
||||
case FeedType.Filter:
|
||||
return {
|
||||
relays: [],
|
||||
filters: (feed as DynamicFilter[]).map(filter => this._compileFilter(filter)),
|
||||
}
|
||||
case FeedType.List:
|
||||
return await this._compileLists(feed as string[])
|
||||
case FeedType.DVM:
|
||||
return await this._compileDvms(feed as DVMItem[])
|
||||
case FeedType.Address: return this._compileAddresses(feed as string[])
|
||||
case FeedType.Author: return this._compileFilter("authors", feed as string[])
|
||||
case FeedType.DVM: return await this._compileDvms(feed as DVMItem[])
|
||||
case FeedType.ID: return this._compileFilter("ids", feed as string[])
|
||||
case FeedType.Intersection: return await this._compileIntersection(feed as Feed[])
|
||||
case FeedType.Kind: return this._compileFilter("kinds", feed as number[])
|
||||
case FeedType.List: return await this._compileLists(feed as ListItem[])
|
||||
case FeedType.Relay: return [{relays: feed as string[]}]
|
||||
case FeedType.Scope: return this._compileScopes(feed as Scope[])
|
||||
case FeedType.Since: return this._compileFilter("since", feed[0] as number)
|
||||
case FeedType.SinceAgo: return this._compileFilter("since", now() - (feed[0] as number))
|
||||
case FeedType.ATag: return this._compileFilter("#a", feed as string[])
|
||||
case FeedType.ETag: return this._compileFilter("#e", feed as string[])
|
||||
case FeedType.PTag: return this._compileFilter("#p", feed as string[])
|
||||
case FeedType.TTag: return this._compileFilter("#t", feed as string[])
|
||||
case FeedType.Until: return this._compileFilter("until", feed[0] as number)
|
||||
case FeedType.UntilAgo: return this._compileFilter("until", now() - (feed[0] as number))
|
||||
case FeedType.Union: return await this._compileUnion(feed as Feed[])
|
||||
case FeedType.WOT: return this._compileWot(feed[0] as number, feed[1] as number)
|
||||
default:
|
||||
throw new Error(`Unable to convert feed of type ${type} to filters`)
|
||||
}
|
||||
}
|
||||
|
||||
async _compileUnion(feeds: Feed[]): Promise<RequestItem> {
|
||||
const relays: string[] = []
|
||||
_compileAddresses(addresses: string[]) {
|
||||
return [{filters: getIdFilters(addresses)}]
|
||||
}
|
||||
|
||||
_compileFilter(key: string, value: any) {
|
||||
return [{filters: [{[key]: value} as Filter]}]
|
||||
}
|
||||
|
||||
_compileScopes(scopes: Scope[]) {
|
||||
return [{filters: [{authors: uniq(scopes.flatMap(this.options.getPubkeysForScope))}]}]
|
||||
}
|
||||
|
||||
_compileWot(min_wot: number, max_wot: number) {
|
||||
return [{filters: [{authors: this.options.getPubkeysForWotRange(min_wot, max_wot)}]}]
|
||||
}
|
||||
|
||||
async _compileDvms(items: DVMItem[]): Promise<RequestItem[]> {
|
||||
const filters: Filter[] = []
|
||||
|
||||
await Promise.all(
|
||||
feeds.map(async feed => {
|
||||
const item = await this.compile(feed)
|
||||
|
||||
for (const relay of item.relays) {
|
||||
relays.push(relay)
|
||||
}
|
||||
|
||||
for (const filter of item.filters) {
|
||||
filters.push(filter)
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
return {
|
||||
relays: uniq(relays),
|
||||
filters: unionFilters(filters),
|
||||
}
|
||||
}
|
||||
|
||||
async _compileIntersection(feeds: Feed[]): Promise<RequestItem> {
|
||||
const items = await Promise.all(feeds.map(this.compile))
|
||||
const filters = intersectFilters(items.map(item => item.filters))
|
||||
|
||||
let relays = uniq(items.flatMap(item => item.relays))
|
||||
let hasRelays = relays.length > 0
|
||||
|
||||
items.forEach((item, i) => {
|
||||
if (item.relays.length > 0) {
|
||||
relays = relays.filter(relay => item.relays.includes(relay))
|
||||
}
|
||||
})
|
||||
|
||||
if (hasRelays && relays.length === 0) {
|
||||
relays.push(BOGUS_RELAY_URL)
|
||||
}
|
||||
|
||||
return {relays, filters}
|
||||
}
|
||||
|
||||
async _compileLists(addresses: string[]): Promise<RequestItem> {
|
||||
const events: E[] = []
|
||||
|
||||
await this.options.request({
|
||||
relays: [],
|
||||
filters: getIdFilters(addresses),
|
||||
onEvent: (e: E) => events.push(e),
|
||||
})
|
||||
|
||||
|
||||
return {
|
||||
relays: [],
|
||||
filters: this._getFiltersFromTags(Tags.fromEvents(events)),
|
||||
}
|
||||
}
|
||||
|
||||
async _compileDvms(requests: DVMItem[]): Promise<RequestItem> {
|
||||
const responseTags: Tags[] = []
|
||||
|
||||
await Promise.all(
|
||||
requests.map(request =>
|
||||
items.map(({mappings, ...request}) =>
|
||||
this.options.requestDvm({
|
||||
...request,
|
||||
onEvent: async (e: E) => {
|
||||
const tags = Tags.fromEvent(e)
|
||||
const {id, pubkey} = await tryCatch(() => JSON.parse(tags.get("request")?.value())) || {}
|
||||
const request = await tryCatch(() => JSON.parse(tags.get("request")?.value()))
|
||||
const responseTags = tags.rejectByValue([request?.id, request?.pubkey])
|
||||
|
||||
responseTags.push(tags.filterByKey(["t", "p", "e", "a"]).rejectByValue([id, pubkey]))
|
||||
for (const filter of await this._getFiltersFromTags(responseTags, mappings)) {
|
||||
filters.push(filter)
|
||||
}
|
||||
},
|
||||
})
|
||||
)
|
||||
)
|
||||
|
||||
const mergedTags = Tags.from(responseTags.flatMap(tags => tags.valueOf()))
|
||||
return [{filters: unionFilters(filters)}]
|
||||
}
|
||||
|
||||
return {
|
||||
relays: mergedTags.relays().valueOf(),
|
||||
filters: this._getFiltersFromTags(mergedTags),
|
||||
async _compileIntersection(feeds: Feed[]): Promise<RequestItem[]> {
|
||||
const [head, ...tail] = await Promise.all(feeds.map(this.compile))
|
||||
|
||||
const result = []
|
||||
|
||||
for (let {filters, relays} of head) {
|
||||
const matchingGroups = tail.map(
|
||||
items => items.filter(
|
||||
it => (
|
||||
(!relays || !it.relays || intersection(relays, it.relays).length > 0) &&
|
||||
(!filters || !it.filters || intersectFilters([filters, it.filters]).length > 0)
|
||||
)
|
||||
)
|
||||
).filter(
|
||||
items => items.length > 0
|
||||
)
|
||||
|
||||
if (matchingGroups.length < tail.length) {
|
||||
continue
|
||||
}
|
||||
|
||||
for (const items of matchingGroups) {
|
||||
for (const item of items) {
|
||||
if (relays && item.relays) {
|
||||
relays = relays.filter(r => item.relays!.includes(r))
|
||||
} else if (item.relays) {
|
||||
relays = item.relays
|
||||
}
|
||||
|
||||
if (filters && item.filters) {
|
||||
filters = intersectFilters([filters, item.filters])
|
||||
} else if (item.filters) {
|
||||
filters = item.filters
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result.push({relays, filters})
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
async _compileUnion(feeds: Feed[]): Promise<RequestItem[]> {
|
||||
const filtersByRelay = new Map<string, Filter[]>()
|
||||
const filtersWithoutRelay: Filter[] = []
|
||||
const relaysWithoutFilter: string[] = []
|
||||
|
||||
await Promise.all(
|
||||
feeds.map(async feed => {
|
||||
for (const item of await this.compile(feed)) {
|
||||
if (item.relays) {
|
||||
for (const relay of item.relays) {
|
||||
if (item.filters) {
|
||||
for (const filter of item.filters) {
|
||||
pushToMapKey(filtersByRelay, relay, filter)
|
||||
}
|
||||
} else {
|
||||
relaysWithoutFilter.push(relay)
|
||||
}
|
||||
}
|
||||
} else if (item.filters) {
|
||||
for (const filter of item.filters) {
|
||||
filtersWithoutRelay.push(filter)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
const items: RequestItem[] = []
|
||||
|
||||
for (const [relay, filters] of filtersByRelay.entries()) {
|
||||
items.push({
|
||||
relays: [relay],
|
||||
filters: unionFilters(filters),
|
||||
})
|
||||
}
|
||||
|
||||
if (filtersWithoutRelay.length > 0) {
|
||||
items.push({filters: unionFilters(filtersWithoutRelay)})
|
||||
}
|
||||
|
||||
if (relaysWithoutFilter.length > 0) {
|
||||
items.push({relays: uniq(relaysWithoutFilter)})
|
||||
}
|
||||
|
||||
return items
|
||||
}
|
||||
|
||||
async _compileLists(listItems: ListItem[]): Promise<RequestItem[]> {
|
||||
const addresses = uniq(listItems.map(({address}) => address))
|
||||
const eventsByAddress = new Map<string, E>()
|
||||
|
||||
await this.options.request({
|
||||
filters: getIdFilters(addresses),
|
||||
onEvent: (e: E) => eventsByAddress.set(getAddress(e), e),
|
||||
})
|
||||
|
||||
const filters = flatten(
|
||||
await Promise.all(
|
||||
listItems.map(({address, mappings}) => {
|
||||
const event = eventsByAddress.get(address)
|
||||
|
||||
return event ? this._getFiltersFromTags(Tags.fromEvent(event), mappings) : []
|
||||
})
|
||||
)
|
||||
)
|
||||
|
||||
return [{filters: unionFilters(filters)}]
|
||||
}
|
||||
|
||||
// Utilities
|
||||
|
||||
_compileFilter({scopes, min_wot, max_wot, until_ago, since_ago, ...filter}: DynamicFilter) {
|
||||
if (scopes && !filter.authors) {
|
||||
filter.authors = scopes.flatMap((scope: Scope) => this.options.getPubkeysForScope(scope))
|
||||
}
|
||||
async _getFiltersFromTags(tags: Tags, mappings: TagFilterMapping[]) {
|
||||
const filters = []
|
||||
|
||||
if ((!isNil(min_wot) || !isNil(max_wot))) {
|
||||
const authors = this.options.getPubkeysForWotRange(min_wot || 0, max_wot || 1)
|
||||
for (const [tagName, feedType] of mappings) {
|
||||
const filterTags = tags.whereKey(tagName)
|
||||
|
||||
if (filter.authors) {
|
||||
const authorsSet = new Set(authors)
|
||||
if (filterTags.exists()) {
|
||||
let values: string[] | number[] = filterTags.values().valueOf()
|
||||
|
||||
filter.authors = filter.authors.filter(pubkey => authorsSet.has(pubkey))
|
||||
} else {
|
||||
filter.authors = authors
|
||||
if (feedType === FeedType.Kind) {
|
||||
values = values.map(ensureNumber) as number[]
|
||||
}
|
||||
|
||||
for (const item of await this.compile([feedType, ...values] as Feed)) {
|
||||
for (const filter of item.filters || []) {
|
||||
filters.push(filter)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!isNil(until_ago)) {
|
||||
filter.until = now() - until_ago!
|
||||
}
|
||||
|
||||
if (!isNil(since_ago)) {
|
||||
filter.since = now() - since_ago!
|
||||
}
|
||||
|
||||
return filter as Filter
|
||||
}
|
||||
|
||||
_getFiltersFromTags(tags: Tags) {
|
||||
const ttags = tags.values("t")
|
||||
const ptags = tags.values("p")
|
||||
const eatags = tags.filterByKey(["e", "a"]).values()
|
||||
const filters: Filter[] = []
|
||||
|
||||
if (ttags.exists()) {
|
||||
filters.push({"#t": ttags.valueOf()})
|
||||
}
|
||||
|
||||
if (ptags.exists()) {
|
||||
filters.push({authors: ptags.valueOf()})
|
||||
}
|
||||
|
||||
if (eatags.exists()) {
|
||||
for (const filter of getIdFilters(eatags.valueOf())) {
|
||||
filters.push(filter)
|
||||
}
|
||||
}
|
||||
|
||||
// If we don't have any filters, return nothing instead of everything
|
||||
if (filters.length === 0) {
|
||||
filters.push({authors: []})
|
||||
}
|
||||
|
||||
return filters
|
||||
return unionFilters(filters)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user