diff --git a/src/app/feeds.ts b/src/app/feeds.ts index 0272ad5d..7134e97e 100644 --- a/src/app/feeds.ts +++ b/src/app/feeds.ts @@ -1,5 +1,5 @@ import {derived, get, readable, writable} from "svelte/store" -import type {Readable, Writable} from "svelte/store" +import type {Readable} from "svelte/store" import {batch, call, int, ms, now, on, sleep, uniqBy, MONTH, YEAR} from "@welshman/lib" import { COMMENT, @@ -298,16 +298,14 @@ export type FeedContext = ReturnType const syncFeed = ({ relays, filters, - events, - seen, - insertEvents, + addEvents, + removeEvents, requireRelay = true, }: { relays: string[] filters: Filter[] - events: Writable - seen: Set - insertEvents: (events: TrustedEvent[]) => void + addEvents: (events: TrustedEvent[]) => void + removeEvents: (ids: Set) => void // Whether an event has to have been seen on one of `relays` to belong to this feed requireRelay?: boolean }) => { @@ -323,7 +321,7 @@ const syncFeed = ({ } if (matching.length > 0) { - insertEvents(matching) + addEvents(matching) } }) @@ -335,11 +333,7 @@ const syncFeed = ({ const {added, removed} = mergeRepositoryUpdates(updates) if (removed.size > 0) { - events.update($events => $events.filter(e => !removed.has(e.id))) - - for (const id of removed) { - seen.delete(id) - } + removeEvents(removed) } const matching = added.filter( @@ -349,7 +343,7 @@ const syncFeed = ({ ) if (matching.length > 0) { - insertEvents(matching) + addEvents(matching) } }), ), @@ -379,6 +373,13 @@ export type FeedSpan = {found: number; complete: boolean; exhausted: boolean} // history on a single request. const SPANS_PER_TRIGGER = 3 +// How many events one step back through the history asks for. Spans are a month wide so that a +// quiet feed finds its history in a few requests, which in a busy one is thousands of events — +// far more than anyone is about to read, and all of it queued ahead of what they are looking at. +// Asking for a page instead leaves the relays to answer with the events nearest the anchor, +// which is what NIP-01 promises a filter carrying a limit. +const PAGE_SIZE = 100 + // Whether a request is actually in flight, which is not the same as whether more might exist. A // list that already has what it needs shouldn't sit under a spinner just because it hasn't // walked to the end of the history. @@ -472,17 +473,27 @@ export const makeFeed = ({ const events = writable([]) const seen = new Set() + // Events from further back than the feed has reached. Everything the app does fills the same + // repository — a space-wide sync reconciling a month of every room at once, most of all — and + // putting each of those on screen as it lands is what walks a room backwards under the reader. + const held = new Map() + // The span the relays have been asked about, which grows outward from the anchor let oldest = at let newest = at let interval = int(MONTH) + // How far back the feed has been answered for, which is what decides whether an event is ready + // to render or has to wait for the window to come and get it. + let reached = at + const insertEvents = (newEvents: Iterable) => { const added: TrustedEvent[] = [] for (const event of newEvents) { if (!seen.has(event.id)) { seen.add(event.id) + held.delete(event.id) added.push(event) onEvent?.(event) } @@ -495,16 +506,79 @@ export const makeFeed = ({ } } - const unsubscribers = syncFeed({relays, filters, events, seen, insertEvents}) + // What arrives from elsewhere in the app, which is only rendered as far back as the feed has + // got to on its own + const addEvents = (newEvents: TrustedEvent[]) => { + const ready: TrustedEvent[] = [] - const loadTimeframe = async (since: number, until: number) => { + for (const event of newEvents) { + if (!seen.has(event.id) && !held.has(event.id)) { + if (event.created_at >= reached) { + ready.push(event) + } else { + held.set(event.id, event) + } + } + } + + insertEvents(ready) + } + + const removeEvents = (ids: Set) => { + events.update($events => $events.filter(event => !ids.has(event.id))) + + for (const id of ids) { + seen.delete(id) + held.delete(id) + } + } + + // Take the feed back to `timestamp`, releasing everything that had arrived for the stretch + // between there and where it had got to + const reach = (timestamp: number) => { + if (timestamp < reached) { + reached = timestamp + + const ready: TrustedEvent[] = [] + + for (const [id, event] of held) { + if (event.created_at >= reached) { + held.delete(id) + ready.push(event) + } + } + + insertEvents(ready) + } + } + + const unsubscribers = syncFeed({relays, filters, addEvents, removeEvents}) + + // One request per direction, reported per relay as well as in total: each relay answers a + // limit for itself, so a page only runs out where the relay that gave the least of it ran out. + const loadSpan = async (extension: Filter) => { let complete = false + const pages = new Map() + + const countEvent = (event: TrustedEvent, url: string) => { + const page = pages.get(url) + + if (page) { + page.count += 1 + page.lowest = Math.min(page.lowest, event.created_at) + } else { + pages.set(url, {count: 1, lowest: event.created_at}) + } + } + const found = await network.get().request({ relays, autoClose: true, signal: controller.signal, - filters: filters.map(filter => ({...filter, since, until})), + filters: filters.map(filter => ({...filter, ...extension})), + onEvent: countEvent, + onDuplicate: countEvent, onEose: () => { complete = true }, @@ -516,7 +590,7 @@ export const makeFeed = ({ interval = found.length > 0 ? int(MONTH) : Math.round(interval * 1.5) } - return {found: found.length, complete} + return {found, complete, pages} } // Ask for the next span in each direction. A span that comes back empty is normal while @@ -529,33 +603,58 @@ export const makeFeed = ({ const until = oldest const since = until - interval - const {found, complete} = await loadTimeframe(since, until) + const {found, complete, pages} = await loadSpan({since, until, limit: PAGE_SIZE}) - if (complete) { - oldest = since + // A relay that answered with less than it was allowed has covered its whole span and holds + // nothing back, so the page runs out at the highest of the rest + let edge: Maybe + + for (const page of pages.values()) { + if (page.count >= PAGE_SIZE && (edge === undefined || page.lowest > edge)) { + edge = page.lowest + } } - return {found, complete, exhausted: false} + if (complete) { + // The second the edge steps back is what stops a page that filled up inside one from being + // asked for over and over + oldest = edge === undefined ? since : Math.min(edge, until - 1) + } + + insertEvents(found) + reach(oldest) + + return {found: found.length, complete, exhausted: false} } + // A limit is answered with the newest events matching it, which reaches away from an anchor in + // the past rather than toward it, so this direction walks spans as it always has const loadNewer = async (): Promise => { if (newest >= now()) return {found: 0, complete: true, exhausted: true} const since = newest const until = Math.min(now(), since + interval) - const {found, complete} = await loadTimeframe(since, until) + const {found, complete} = await loadSpan({since, until}) if (complete) { newest = until } - return {found, complete, exhausted: false} + insertEvents(found) + + return {found: found.length, complete, exhausted: false} } - for (const url of relays) { - insertEvents(getEventsForUrl(url, filters)) + // What the repository already holds for these relays is in hand and goes in as one insert, + // which takes the window back with it rather than leaving the rest of that stretch behind + const cached = relays.flatMap(url => Array.from(getEventsForUrl(url, filters))) + + for (const event of cached) { + reached = Math.min(reached, event.created_at) } + insertEvents(cached) + return { events, loadOlder, @@ -624,14 +723,21 @@ export const makeCalendarFeed = ({ }) } + const removeEvents = (ids: Set) => { + events.update($events => $events.filter(event => !ids.has(event.id))) + + for (const id of ids) { + seen.delete(id) + } + } + // Calendar events are addressable and often relayed on from elsewhere, so this feed takes any // matching event rather than only those seen on its own relays const unsubscribers = syncFeed({ relays, filters, - events, - seen, - insertEvents, + addEvents: insertEvents, + removeEvents, requireRelay: false, })