import {derived, get, readable, writable} from "svelte/store" import type {Readable, Writable} from "svelte/store" import {batch, call, int, ms, now, on, sleep, sortBy, uniqBy, MONTH, YEAR} from "@welshman/lib" import { COMMENT, DELETE, EVENT_TIME, addressTags, getAddress, getCommentFiltersForRoot, getIdOrAddress, getReplyFilters, hexTags, isReplaceableKind, matchFilters, tagSpec, tagValue, tagValues, } from "@welshman/util" import type {Maybe} from "@welshman/lib" import type {Filter, TrustedEvent} from "@welshman/util" import {mergeRepositoryUpdates} from "@welshman/net" import type {RepositoryUpdate} from "@welshman/net" import {createScroller} from "@lib/html" import type {ScrollerOpts} from "@lib/html" import {daysBetween} from "@lib/util" import {EVENT_CONTEXT_KINDS, REACTION_KINDS} from "@app/content" import {app, network} from "@app/core" import {getEventsForUrl} from "@app/repository" const noEvents: TrustedEvent[] = [] // Reactions, zaps and reports point at their subject with `e`/`a`. A NIP-22 comment instead // points at its thread *root* with `E`/`A`, so filing it by those tags puts a whole thread in // the root's bucket — which is the scope a reply count wants. const getTargets = ({kind, tags}: TrustedEvent) => kind === COMMENT ? [...tagValues(hexTags("E"), tags), ...tagValues(addressTags("A"), tags)] : [...tagValues(hexTags("e"), tags), ...tagValues(addressTags("a"), tags)] // A related event may point at a replaceable one by either id or address, so both are keys const getKeys = (event: TrustedEvent) => isReplaceableKind(event.kind) ? [event.id, getAddress(event)] : [event.id] export const makeFeedContext = ({relays}: {relays: string[] | Promise}) => { const {repository} = app.get() const controller = new AbortController() const targets = new Set() const eventsByTarget = new Map() const targetsByEventId = new Map() const subscribersByTarget = new Map void>>() const deletedChecks = new Map void>>() const addEvent = (event: TrustedEvent, touched: Set) => { // An event seen before its target was tracked stays unfiled, so that adding the target // later can pick it up out of the repository if (!targetsByEventId.has(event.id)) { const eventTargets = getTargets(event).filter(target => targets.has(target)) if (eventTargets.length > 0) { targetsByEventId.set(event.id, eventTargets) for (const target of eventTargets) { eventsByTarget.set(target, [...(eventsByTarget.get(target) || []), event]) touched.add(target) } } } } const removeEvent = (id: string, touched: Set) => { const eventTargets = targetsByEventId.get(id) if (eventTargets) { targetsByEventId.delete(id) for (const target of eventTargets) { const events = eventsByTarget.get(target)!.filter(event => event.id !== id) if (events.length > 0) { eventsByTarget.set(target, events) } else { eventsByTarget.delete(target) } touched.add(target) } } } const notifyDeleted = (added: TrustedEvent[], removed: Set) => { if (deletedChecks.size > 0) { if (removed.size > 0) { for (const checks of deletedChecks.values()) { for (const check of checks) check() } } else { for (const event of added) { for (const check of deletedChecks.get(getIdOrAddress(event)) || []) { check() } } } } } const notify = (touched: Set) => { for (const target of touched) { for (const subscriber of subscribersByTarget.get(target) || []) { subscriber(eventsByTarget.get(target) || noEvents) } } } const loadContext = batch(100, async (events: TrustedEvent[]) => { const touched = new Set() // What's already local — an earlier page, our own optimistic reactions — never comes // through the update listener, so file it before asking the network for the rest for (const event of repository.query([ ...getReplyFilters(events, {kinds: EVENT_CONTEXT_KINDS}), ...getCommentFiltersForRoot(events), ])) { addEvent(event, touched) } notify(touched) const urls = await relays const context = await network.get().load({ relays: urls, signal: controller.signal, filters: [ ...getReplyFilters(events, {kinds: REACTION_KINDS}), ...getCommentFiltersForRoot(events), ], }) if (context.length > 0) { network.get().load({ relays: urls, signal: controller.signal, filters: getReplyFilters(context, {kinds: [DELETE]}), }) } }) const unsubscribe = on( repository, "update", batch(150, (updates: RepositoryUpdate[]) => { const {added, removed} = mergeRepositoryUpdates(updates) const touched = new Set() for (const event of added) { if (EVENT_CONTEXT_KINDS.includes(event.kind)) { addEvent(event, touched) } } for (const id of removed) { removeEvent(id, touched) } notify(touched) notifyDeleted(added, removed) }), ) // Track an event so its context gets requested with the rest of the batch const add = (event: TrustedEvent) => { if (!targets.has(event.id)) { for (const key of getKeys(event)) { targets.add(key) } loadContext(event) } } const relatedForKey = (key: string) => readable(eventsByTarget.get(key) || noEvents, set => { let subscribers = subscribersByTarget.get(key) if (!subscribers) { subscribers = new Set() subscribersByTarget.set(key, subscribers) } subscribers.add(set) set(eventsByTarget.get(key) || noEvents) return () => { subscribers.delete(set) if (subscribers.size === 0) { subscribersByTarget.delete(key) } } }) return { add, deleted: (event: TrustedEvent) => readable(repository.isDeleted(event), set => { const key = getIdOrAddress(event) const check = () => set(repository.isDeleted(event)) let checks = deletedChecks.get(key) if (!checks) { checks = new Set() deletedChecks.set(key, checks) } checks.add(check) check() return () => { checks.delete(check) if (checks.size === 0) { deletedChecks.delete(key) } } }), related: (event: TrustedEvent): Readable => { add(event) const [key, ...rest] = getKeys(event) // A replaceable event collects both its buckets, and something tagging it by id and // address at once lands in both return rest.length > 0 ? derived([relatedForKey(key), ...rest.map(relatedForKey)], buckets => uniqBy(event => event.id, buckets.flat()), ) : relatedForKey(key) }, cleanup: () => { controller.abort() unsubscribe() }, } } export type FeedContext = ReturnType // Keeps a feed's store in step with the repository: events that arrive later, events that get // deleted, and events already held that have only now been seen on one of the feed's relays. const syncFeed = ({ relays, filters, events, seen, insertEvents, requireRelay = true, }: { relays: string[] filters: Filter[] events: Writable seen: Set insertEvents: (events: TrustedEvent[]) => void // Whether an event has to have been seen on one of `relays` to belong to this feed requireRelay?: boolean }) => { const onTrackedId = batch(150, (ids: string[]) => { const matching: TrustedEvent[] = [] for (const id of new Set(ids)) { const event = app.get().repository.getEvent(id) if (event && matchFilters(filters, event)) { matching.push(event) } } if (matching.length > 0) { insertEvents(matching) } }) return [ on( app.get().repository, "update", batch(150, (updates: RepositoryUpdate[]) => { 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) } } const matching = added.filter( event => matchFilters(filters, event) && (!requireRelay || relays.some(url => app.get().tracker.getRelays(event.id).has(url))), ) if (matching.length > 0) { insertEvents(matching) } }), ), on(app.get().tracker, "add", (id: string, url: string) => { if (relays.includes(url)) { onTrackedId(id) } }), ] } // One direction of a feed. A span that comes back empty is a gap in the timeline, not the end of // it — conflating the two either stops loading at the first gap or walks the whole history // looking for the end of one. export type FeedLoadState = | {status: "idle"} | {status: "loading"} | {status: "searching"} | {status: "exhausted"} // One span of a feed's timeline. `complete` is whether the relays actually answered for it: a // request the socket dropped reports nothing found, which is not the same thing as a span with // nothing in it, and the two have to move the window differently. export type FeedSpan = {found: number; complete: boolean; exhausted: boolean} // Empty spans to walk per trigger. Enough to cross a gap; not enough to reach the end of the // history on a single request. const SPANS_PER_TRIGGER = 3 // 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. export const isFeedLoading = (state: Maybe) => state?.status === "loading" || state?.status === "searching" const makeFeedLoader = (load: () => Promise) => { const state = writable({status: "idle"}) let running = false const run = async () => { if (running || get(state).status === "exhausted") return running = true try { for (let span = 0; span < SPANS_PER_TRIGGER; span++) { state.set({status: span > 0 ? "searching" : "loading"}) const {found, complete, exhausted} = await load() if (exhausted) { state.set({status: "exhausted"}) return } // A span nobody answered for hasn't moved the window, so hold here rather than walking // past it, and give the socket a moment before the scroller comes back around if (!complete) { await sleep(ms(3)) break } if (found > 0) break } state.set({status: "idle"}) } finally { running = false } } return {subscribe: state.subscribe, run} } // A loader triggered by proximity to the end of a scroll container, which is how every list in // the app pages. The container's orientation decides which direction `reverse` reaches, so the // caller passes it — a reversed chat scrolls away from its origin to find older messages, an // ordinary feed scrolls toward the end of its own content. export const makeScrollLoader = ( element: HTMLElement, load: () => Promise, options: Partial = {}, ) => { const loader = makeFeedLoader(load) const scroller = createScroller({ element, delay: 300, threshold: 5000, ...options, onScroll: loader.run, }) return {subscribe: loader.subscribe, stop: scroller.stop} } // Holds every event a view has loaded, sorted oldest to newest, and knows how to ask for the // next span in either direction. It does not decide *when* to ask — the view does, because the // view is what knows what is on screen. export const makeFeed = ({ relays, filters, onEvent, at = now(), }: { relays: string[] filters: Filter[] onEvent?: (event: TrustedEvent) => void at?: number }) => { const controller = new AbortController() const events = writable([]) const seen = new Set() // The span the relays have been asked about, which grows outward from the anchor let oldest = at let newest = at let interval = int(MONTH) const insertEvents = (newEvents: Iterable) => { const added: TrustedEvent[] = [] for (const event of newEvents) { if (!seen.has(event.id)) { seen.add(event.id) added.push(event) onEvent?.(event) } } if (added.length > 0) { added.sort((a, b) => a.created_at - b.created_at) events.update($events => { const merged: TrustedEvent[] = [] let i = 0 let j = 0 while (i < $events.length && j < added.length) { if ($events[i].created_at <= added[j].created_at) { merged.push($events[i++]) } else { merged.push(added[j++]) } } while (i < $events.length) merged.push($events[i++]) while (j < added.length) merged.push(added[j++]) return merged }) } } const unsubscribers = syncFeed({relays, filters, events, seen, insertEvents}) const loadTimeframe = async (since: number, until: number) => { let complete = false const found = await network.get().request({ relays, autoClose: true, signal: controller.signal, filters: filters.map(filter => ({...filter, since, until})), onEose: () => { complete = true }, }) // A span that turns up nothing widens the next one, so walking a sparse history doesn't // take dozens of round trips if (complete) { interval = found.length > 0 ? int(MONTH) : Math.round(interval * 1.5) } return {found: found.length, complete} } // Ask for the next span in each direction. A span that comes back empty is normal while // walking a sparse history, so the count is reported separately from whether there is any // history left — a caller watching its list for changes would never hear about an empty one. // The window only moves once the relays have answered: a request the socket dropped looks // exactly like an empty span, and walking past it would leave a hole nothing goes back for. const loadOlder = async (): Promise => { if (oldest < now() - int(2, YEAR)) return {found: 0, complete: true, exhausted: true} const until = oldest const since = until - interval const {found, complete} = await loadTimeframe(since, until) if (complete) { oldest = since } return {found, complete, exhausted: false} } 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) if (complete) { newest = until } return {found, complete, exhausted: false} } for (const url of relays) { insertEvents(getEventsForUrl(url, filters)) } return { events, loadOlder, loadNewer, cleanup: () => { controller.abort() unsubscribers.forEach(call) }, } } // Same split as makeFeed: it holds what has been loaded and knows how to reach further out in // either direction, while the page decides when to ask. Calendar events are addressed by the // days they cover rather than by when they were published, so the spans are date hashes. export const makeCalendarFeed = ({ relays, filters, onEvent, }: { relays: string[] filters: Filter[] onEvent?: (event: TrustedEvent) => void }) => { const interval = int(5, MONTH) const controller = new AbortController() const seen = new Set() let oldest = now() let newest = now() const getStart = (event: TrustedEvent) => parseInt(tagValue(tagSpec("start"), event.tags) || "") const getEnd = (event: TrustedEvent) => parseInt(tagValue(tagSpec("end"), event.tags) || "") const events = writable( sortBy( getStart, uniqBy( e => e.id, relays.flatMap(url => Array.from(getEventsForUrl(url, filters))), ), ), ) const insertEvents = (newEvents: TrustedEvent[]) => { const valid = newEvents.filter(e => !isNaN(getStart(e)) && !isNaN(getEnd(e)) && !seen.has(e.id)) if (valid.length === 0) return for (const event of valid) { seen.add(event.id) onEvent?.(event) } valid.sort((a, b) => getStart(a) - getStart(b)) events.update($events => { // Calendar events are addressable, so a new version supersedes the old one const superseded = new Set(valid.map(getAddress)) const kept = $events.filter(e => !superseded.has(getAddress(e))) const merged: TrustedEvent[] = [] let i = 0 let j = 0 while (i < kept.length && j < valid.length) { if (getStart(kept[i]) <= getStart(valid[j])) { merged.push(kept[i++]) } else { merged.push(valid[j++]) } } while (i < kept.length) merged.push(kept[i++]) while (j < valid.length) merged.push(valid[j++]) return merged }) } // 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, requireRelay: false, }) const loadTimeframe = async (since: number, until: number) => { let complete = false const found = await network.get().request({ relays, autoClose: true, signal: controller.signal, filters: [{kinds: [EVENT_TIME], "#D": daysBetween(since, until).map(String)}], onEose: () => { complete = true }, }) return {found: found.length, complete} } const loadOlder = async (): Promise => { if (oldest < now() - int(2, YEAR)) return {found: 0, complete: true, exhausted: true} const until = oldest const since = until - interval const {found, complete} = await loadTimeframe(since, until) if (complete) { oldest = since } return {found, complete, exhausted: false} } const loadNewer = async (): Promise => { if (newest > now() + int(2, YEAR)) return {found: 0, complete: true, exhausted: true} const since = newest const until = since + interval const {found, complete} = await loadTimeframe(since, until) if (complete) { newest = until } return {found, complete, exhausted: false} } return { events, loadOlder, loadNewer, cleanup: () => { controller.abort() unsubscribers.forEach(call) }, } }