coracle/welshman
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move feed load opts to load

Browse Source
This commit is contained in:
Jon Staab
2024-04-15 09:26:06 -07:00
parent 2e54f9040d
commit bd08c2c283
4 changed files with 134 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/feeds/compiler.ts
+75 -31
View File
@@ -1,31 +1,36 @@
import {inc, now, isNil} from '@coracle.social/lib'
import {inc, uniq, now, isNil} from '@coracle.social/lib'
import type {Rumor, Filter} from '@coracle.social/util'
import {Tags, getIdFilters, mergeFilters} from '@coracle.social/util'
import type {DVMRequest, Scope, Feed, DynamicFilter} from './core'
import type {RequestItem, DVMItem, Scope, Feed, DynamicFilter, FeedContext} from './core'
import {FeedType} from './core'
export type ExecuteOpts<E> = {
onEvent: (event: E) => void
}
export type FeedCompilerOpts<E> = {
reqDvm: (request: DVMRequest, opts: ExecuteOpts<E>) => Promise<void>
reqFilters: (filters: Filter[], opts: ExecuteOpts<E>) => Promise<void>
getPubkeysForScope: (scope: Scope) => string[]
getPubkeysForWotRange: (minWot: number, maxWot: number) => string[]
}
export class FeedCompiler<E extends Rumor> {
constructor(readonly opts: FeedCompilerOpts<E>) {}
constructor(readonly context: FeedContext<E>) {}
walk([type, ...feed]: Feed, visit: (feed: Feed) => void) {
visit([type, ...feed] as Feed)
switch(type) {
case FeedType.Difference:
case FeedType.Intersection:
case FeedType.SymmetricDifference:
case FeedType.Union:
for (const subFeed of feed) {
this.walk(subFeed as Feed, visit)
}
}
}
canCompile([type, ...feed]: Feed): boolean {
switch(type) {
case FeedType.Relay:
return (feed.slice(1) as Feed[]).every(this.canCompile)
case FeedType.Union:
return (feed as Feed[]).every(this.canCompile)
case FeedType.Filter:
case FeedType.List:
case FeedType.LOL:
case FeedType.DVM:
case FeedType.Filter:
return true
default:
return false
@@ -36,66 +41,105 @@ export class FeedCompiler<E extends Rumor> {
switch(type) {
case FeedType.Union:
return await this._compileUnion(feed as Feed[])
case FeedType.Relay:
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.LOL:
return await this._compileLols(feed as string[])
case FeedType.DVM:
return await this._compileDvms(feed as DVMRequest[])
case FeedType.Filter:
return (feed as DynamicFilter[]).map(filter => this._compileFilter(filter))
return await this._compileDvms(feed as DVMItem[])
default:
throw new Error(`Unable to convert feed of type ${type} to filters`)
}
}
async _compileUnion(feeds: Feed[]): Promise<Filter[]> {
async _compileUnion(feeds: Feed[]): Promise<RequestItem> {
const relays: string[] = []
const filters: Filter[] = []
await Promise.all(
feeds.map(async feed => {
for (const filter of await this.compile(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 mergeFilters(filters)
return {
relays: uniq(relays),
filters: mergeFilters(filters),
}
}
async _compileLists(addresses: string[]): Promise<Filter[]> {
async _compileLists(addresses: string[]): Promise<RequestItem> {
const events: E[] = []
await this.opts.reqFilters(getIdFilters(addresses), {onEvent: events.push})
await this.context.request({
relays: [],
filters: getIdFilters(addresses),
onEvent: events.push,
})
return this._getFiltersFromTags(Tags.fromEvents(events))
return {
relays: [],
filters: this._getFiltersFromTags(Tags.fromEvents(events)),
}
}
async _compileLols(addresses: string[]): Promise<Filter[]> {
async _compileLols(addresses: string[]): Promise<RequestItem> {
const events: E[] = []
await this.opts.reqFilters(getIdFilters(addresses), {onEvent: events.push})
await this.context.request({
relays: [],
filters: getIdFilters(addresses),
onEvent: events.push,
})
return this._compileLists(Tags.fromEvents(events).values("a").valueOf())
}
async _compileDvms(requests: DVMRequest[]): Promise<Filter[]> {
async _compileDvms(requests: DVMItem[]): Promise<RequestItem> {
const events: E[] = []
await Promise.all(requests.map(request => this.opts.reqDvm(request, {onEvent: events.push})))
await Promise.all(
requests.map(request =>
this.context.requestDvm({
tags: [],
...request,
onEvent: events.push,
})
)
)
return this._getFiltersFromTags(Tags.fromEvents(events))
return {
relays: [],
filters: this._getFiltersFromTags(Tags.fromEvents(events)),
}
}
// Utilities
_compileFilter({scopes, min_wot, max_wot, until_ago, since_ago, ...filter}: DynamicFilter) {
if (scopes && !filter.authors) {
filter.authors = scopes.flatMap((scope: Scope) => this.opts.getPubkeysForScope(scope))
filter.authors = scopes.flatMap((scope: Scope) => this.context.getPubkeysForScope(scope))
}
if ((!isNil(min_wot) || !isNil(max_wot))) {
const authors = this.opts.getPubkeysForWotRange(min_wot || 0, max_wot || 1)
const authors = this.context.getPubkeysForWotRange(min_wot || 0, max_wot || 1)
if (filter.authors) {
const authorsSet = new Set(authors)
packages/feeds/core.ts
+37 -25
View File
@@ -29,22 +29,18 @@ export type DynamicFilter = Filter & {
since_ago?: number
}
export type DVMRequest = {
kind: number
tags?: string[][]
}
export type RelayFeed = [FeedType.Relay, string[], ...Feed[]]
export type DifferenceFeed = [FeedType.Difference, ...Feed[]]
export type IntersectionFeed = [FeedType.Intersection, ...Feed[]]
export type SymmetricDifferenceFeed = [FeedType.SymmetricDifference, ...Feed[]]
export type UnionFeed = [FeedType.Union, ...Feed[]]
export type FilterFeed = [FeedType.Filter, ...DynamicFilter[]]
export type RelayFeed = [FeedType.Relay, ...string[]]
export type ListFeed = [FeedType.List, ...string[]]
export type LOLFeed = [FeedType.LOL, ...string[]]
export type DVMFeed = [FeedType.DVM, ...DVMRequest[]]
export type DVMFeed = [FeedType.DVM, ...DVMItem[]]
export type Feed =
RelayFeed |
DifferenceFeed |
IntersectionFeed |
SymmetricDifferenceFeed |
@@ -55,21 +51,37 @@ export type Feed =
LOLFeed |
DVMFeed
export const difference = (...feeds: Feed[]) =>
[FeedType.Difference, ...feeds] as DifferenceFeed
export const intersection = (...feeds: Feed[]) =>
[FeedType.Intersection, ...feeds] as IntersectionFeed
export const symmetricDifference = (...feeds: Feed[]) =>
[FeedType.SymmetricDifference, ...feeds] as SymmetricDifferenceFeed
export const union = (...feeds: Feed[]) =>
[FeedType.Union, ...feeds] as UnionFeed
export const filter = (...filters: DynamicFilter[]) =>
[FeedType.Filter, ...filters] as FilterFeed
export const relay = (...relays: string[]) =>
[FeedType.Relay, ...relays] as RelayFeed
export const list = (...addresses: string[]) =>
[FeedType.List, ...addresses] as ListFeed
export const lol = (...addresses: string[]) =>
[FeedType.LOL, ...addresses] as LOLFeed
export const dvm = (...requests: DVMRequest[]) =>
[FeedType.DVM, ...requests] as DVMFeed
export const usingRelays = (relays: string[], ...feeds: Feed[]) => [FeedType.Relay, relays, ...feeds] as Feed
export const difference = (...feeds: Feed[]) => [FeedType.Difference, ...feeds] as Feed
export const intersection = (...feeds: Feed[]) => [FeedType.Intersection, ...feeds] as Feed
export const symmetricDifference = (...feeds: Feed[]) => [FeedType.SymmetricDifference, ...feeds] as Feed
export const union = (...feeds: Feed[]) => [FeedType.Union, ...feeds] as Feed
export const filter = (...filters: DynamicFilter[]) => [FeedType.Filter, ...filters] as Feed
export const list = (...addresses: string[]) => [FeedType.List, ...addresses] as Feed
export const lol = (...addresses: string[]) => [FeedType.LOL, ...addresses] as Feed
export const dvm = (...requests: DVMItem[]) => [FeedType.DVM, ...requests] as Feed
export type RequestItem = {
relays: string[]
filters: Filter[]
}
export type RequestOpts<E> = RequestItem & {
onEvent: (event: E) => void
}
export type DVMItem = {
kind: number
tags?: string[][]
}
export type DVMOpts<E> = DVMItem & {
onEvent: (event: E) => void
}
export type FeedContext<E> = {
request: (opts: RequestOpts<E>) => Promise<void>
requestDvm: (opts: DVMOpts<E>) => Promise<void>
getPubkeysForScope: (scope: Scope) => string[]
getPubkeysForWotRange: (minWot: number, maxWot: number) => string[]
}
packages/feeds/index.ts
+1
View File
@@ -1,2 +1,3 @@
export * from './core'
export * from './compiler'
export * from './loader'
packages/feeds/loader.ts
+21 -25
View File
@@ -1,11 +1,11 @@
import {inc, max, min, now, isNil} from '@coracle.social/lib'
import type {Rumor, Filter} from '@coracle.social/util'
import {Tags, EPOCH, getIdFilters, guessFilterDelta, mergeFilters} from '@coracle.social/util'
import type {DVMRequest, Scope, Feed, DynamicFilter} from './core'
import type {Scope, Feed, DynamicFilter, RequestOpts, RequestItem, FeedContext} from './core'
import {FeedType} from './core'
import type {FeedCompiler} from './compiler'
import {FeedCompiler} from './compiler'
export type LoaderOpts<E> = {
export type LoadOpts<E> = {
onEvent: (event: E) => void
onExhausted: () => void
}
@@ -13,38 +13,32 @@ export type LoaderOpts<E> = {
export type Loader = (limit: number) => Promise<void>
export class FeedLoader<E extends Rumor> {
_loader: Promise<Loader>
compiler: FeedCompiler<E>
constructor(readonly compiler: FeedCompiler<E>, feed: Feed, opts: LoaderOpts<E>) {
this._loader = this.getLoader(feed, opts)
constructor(readonly context: FeedContext<E>) {
this.compiler = new FeedCompiler(context)
}
loadMore(limit: number) {
this._loader.then(loader => loader(limit))
}
async getLoader([type, ...feed]: Feed, opts: LoaderOpts<E>) {
async getLoader([type, ...feed]: Feed, loadOpts: LoadOpts<E>) {
if (this.compiler.canCompile([type, ...feed] as Feed)) {
const filters = await this.compiler.compile([type, ...feed] as Feed)
return this._getFilterLoader(filters, opts)
return this._getRequestLoader(await this.compiler.compile([type, ...feed] as Feed), loadOpts)
}
switch(type) {
case FeedType.Difference:
return this._getDifferenceLoader(feed as Feed[], opts)
return this._getDifferenceLoader(feed as Feed[], loadOpts)
case FeedType.Intersection:
return this._getIntersectionLoader(feed as Feed[], opts)
return this._getIntersectionLoader(feed as Feed[], loadOpts)
case FeedType.SymmetricDifference:
return this._getSymmetricDifferenceLoader(feed as Feed[], opts)
return this._getSymmetricDifferenceLoader(feed as Feed[], loadOpts)
case FeedType.Union:
return this._getUnionLoader(feed as Feed[], opts)
return this._getUnionLoader(feed as Feed[], loadOpts)
default:
throw new Error(`Unable to convert feed of type ${type} to loader`)
}
}
async _getFilterLoader(filters: Filter[], {onEvent, onExhausted}: LoaderOpts<E>) {
async _getRequestLoader({relays, filters}: RequestItem, {onEvent, onExhausted}: LoadOpts<E>) {
const untils = filters.flatMap((filter: Filter) => filter.until ? [filter.until] : [])
const sinces = filters.flatMap((filter: Filter) => filter.since ? [filter.since] : [])
const maxUntil = untils.length === filters.length ? max(untils) : now()
@@ -56,7 +50,7 @@ export class FeedLoader<E extends Rumor> {
let until = maxUntil
return async (limit: number) => {
const reqFilters = filters
const requestFilters = filters
// Remove filters that don't fit our window
.filter((filter: Filter) => {
const filterSince = filter.since || EPOCH
@@ -69,7 +63,9 @@ export class FeedLoader<E extends Rumor> {
let count = 0
await this.compiler.opts.reqFilters(filters, {
await this.context.request({
relays,
filters: requestFilters,
onEvent: (event: E) => {
count += 1
since = Math.min(since, event.created_at)
@@ -91,7 +87,7 @@ export class FeedLoader<E extends Rumor> {
}
}
async _getDifferenceLoader(feeds: Feed[], {onEvent, onExhausted}: LoaderOpts<E>) {
async _getDifferenceLoader(feeds: Feed[], {onEvent, onExhausted}: LoadOpts<E>) {
const exhausted = new Set<number>()
const skip = new Set<string>()
const events: E[] = []
@@ -136,7 +132,7 @@ export class FeedLoader<E extends Rumor> {
}
}
async _getIntersectionLoader(feeds: Feed[], {onEvent, onExhausted}: LoaderOpts<E>) {
async _getIntersectionLoader(feeds: Feed[], {onEvent, onExhausted}: LoadOpts<E>) {
const exhausted = new Set<number>()
const counts = new Map<string, number>()
const events: E[] = []
@@ -178,7 +174,7 @@ export class FeedLoader<E extends Rumor> {
}
}
async _getSymmetricDifferenceLoader(feeds: Feed[], {onEvent, onExhausted}: LoaderOpts<E>) {
async _getSymmetricDifferenceLoader(feeds: Feed[], {onEvent, onExhausted}: LoadOpts<E>) {
const exhausted = new Set<number>()
const counts = new Map<string, number>()
const events: E[] = []
@@ -220,7 +216,7 @@ export class FeedLoader<E extends Rumor> {
}
}
async _getUnionLoader(feeds: Feed[], {onEvent, onExhausted}: LoaderOpts<E>) {
async _getUnionLoader(feeds: Feed[], {onEvent, onExhausted}: LoadOpts<E>) {
const exhausted = new Set<number>()
const seen = new Set()