Refactor synchronization logic

2025-10-06 11:23:19 -07:00
parent d0491ed202
commit e0099141aa
17 changed files with 357 additions and 375 deletions
@@ -1,20 +1,18 @@
 import {get, writable} from "svelte/store"
 import {
-  partition,
  uniq,
  int,
  YEAR,
  DAY,
  insertAt,
  sortBy,
-  assoc,
  now,
+  on,
  isNotNil,
  filterVals,
  fromPairs,
 } from "@welshman/lib"
 import {
-  DELETE,
  EVENT_TIME,
  AUTH_INVITE,
  ALERT_EMAIL,
@@ -23,7 +21,6 @@ import {
  ALERT_ANDROID,
  ALERT_STATUS,
  matchFilters,
-  getTagValues,
  getTagValue,
  getAddress,
  isShareableRelayUrl,
@@ -32,74 +29,37 @@ import {
 import type {TrustedEvent, Filter, List} from "@welshman/util"
 import {feedFromFilters, makeRelayFeed, makeIntersectionFeed} from "@welshman/feeds"
 import {load, request} from "@welshman/net"
-import type {AppSyncOpts, Thunk} from "@welshman/app"
-import {
-  repository,
-  pull,
-  hasNegentropy,
-  thunkQueue,
-  makeFeedController,
-  loadRelay,
-} from "@welshman/app"
+import {repository, makeFeedController, loadRelay} from "@welshman/app"
 import {createScroller} from "@lib/html"
 import {daysBetween} from "@lib/util"
-import {NOTIFIER_RELAY, getUrlsForEvent} from "@app/core/state"
+import {NOTIFIER_RELAY, getEventsForUrl} from "@app/core/state"

 // Utils

-export const pullConservatively = ({relays, filters}: AppSyncOpts) => {
-  const $getUrlsForEvent = get(getUrlsForEvent)
-  const [smart, dumb] = partition(hasNegentropy, relays)
-  const promises = [pull({relays: smart, filters})]
-  const allEvents = repository.query(filters, {shouldSort: false})
-
-  // Since pulling from relays without negentropy is expensive, limit how many
-  // duplicates we repeatedly download
-  for (const url of dumb) {
-    const events = allEvents.filter(e => $getUrlsForEvent(e.id).includes(url))
-
-    if (events.length > 100) {
-      filters = filters.map(assoc("since", events[10]!.created_at))
-    }
-
-    promises.push(pull({relays: [url], filters}))
-  }
-
-  return Promise.all(promises)
-}
-
 export const makeFeed = ({
-  relays,
-  feedFilters,
-  subscriptionFilters,
+  url,
+  filters,
  element,
-  onEvent,
  onExhausted,
-  initialEvents = [],
 }: {
-  relays: string[]
-  feedFilters: Filter[]
-  subscriptionFilters: Filter[]
+  url: string
+  filters: Filter[]
  element: HTMLElement
-  onEvent?: (event: TrustedEvent) => void
  onExhausted?: () => void
-  initialEvents?: TrustedEvent[]
 }) => {
-  const seen = new Set<string>()
-  const buffer = writable<TrustedEvent[]>([])
-  const events = writable(initialEvents)
+  const initialEvents = getEventsForUrl(url, filters)
+  const seen = new Set(initialEvents.map(e => e.id))
  const controller = new AbortController()
-
-  const markEvent = (event: TrustedEvent) => {
-    if (!seen.has(event.id)) {
-      seen.add(event.id)
-      onEvent?.(event)
-    }
-  }
+  const buffer = writable(initialEvents)
+  const events = writable<TrustedEvent[]>([])

  const insertEvent = (event: TrustedEvent) => {
    let handled = false

+    if (seen.has(event.id)) {
+      return
+    }
+
    events.update($events => {
      for (let i = 0; i < $events.length; i++) {
        if ($events[i].id === event.id) return $events
@@ -123,47 +83,27 @@ export const makeFeed = ({
      })
    }

-    markEvent(event)
+    seen.add(event.id)
  }

-  const removeEvents = (ids: string[]) => {
-    buffer.update($buffer => $buffer.filter(e => !ids.includes(e.id)))
-    events.update($events => $events.filter(e => !ids.includes(e.id)))
-  }
-
-  const handleDelete = (e: TrustedEvent) => removeEvents(getTagValues(["e", "a"], e.tags))
-
-  const onThunk = (thunk: Thunk) => {
-    if (matchFilters(feedFilters, thunk.event)) {
-      insertEvent(thunk.event)
-
-      thunk.controller.signal.addEventListener("abort", () => {
-        removeEvents([thunk.event.id])
-      })
-    } else if (thunk.event.kind === DELETE) {
-      handleDelete(thunk.event)
+  const unsubscribe = on(repository, "update", ({added, removed}) => {
+    if (removed.size > 0) {
+      buffer.update($buffer => $buffer.filter(e => !removed.has(e.id)))
+      events.update($events => $events.filter(e => !removed.has(e.id)))
    }
-  }
+
+    for (const event of added) {
+      if (matchFilters(filters, event)) {
+        insertEvent(event)
+      }
+    }
+  })

  const ctrl = makeFeedController({
    useWindowing: true,
-    feed: makeIntersectionFeed(makeRelayFeed(...relays), feedFromFilters(feedFilters)),
-    onEvent: insertEvent,
-    onExhausted,
-  })
-
-  for (const event of initialEvents) {
-    markEvent(event)
-  }
-
-  request({
-    relays,
    signal: controller.signal,
-    filters: subscriptionFilters,
-    onEvent: (e: TrustedEvent) => {
-      if (matchFilters(feedFilters, e)) insertEvent(e)
-      if (e.kind === DELETE) handleDelete(e)
-    },
+    feed: makeIntersectionFeed(makeRelayFeed(url), feedFromFilters(filters)),
+    onExhausted,
  })

  const scroller = createScroller({
@@ -173,7 +113,7 @@ export const makeFeed = ({
    onScroll: async () => {
      const $buffer = get(buffer)

-      events.update($events => sortBy(e => -e.created_at, [...$events, ...$buffer.splice(0, 100)]))
+      events.update($events => [...$events, ...$buffer.splice(0, 100)])

      if ($buffer.length < 100) {
        ctrl.load(100)
@@ -181,8 +121,6 @@ export const makeFeed = ({
    },
  })

-  const unsubscribe = thunkQueue.subscribe(onThunk)
-
  return {
    events,
    cleanup: () => {
@@ -194,22 +132,19 @@ export const makeFeed = ({
 }

 export const makeCalendarFeed = ({
-  relays,
-  feedFilters,
-  subscriptionFilters,
+  url,
+  filters,
  element,
  onExhausted,
-  initialEvents = [],
 }: {
-  relays: string[]
-  feedFilters: Filter[]
-  subscriptionFilters: Filter[]
+  url: string
+  filters: Filter[]
  element: HTMLElement
  onExhausted?: () => void
-  initialEvents?: TrustedEvent[]
 }) => {
  const interval = int(5, DAY)
  const controller = new AbortController()
+  const initialEvents = getEventsForUrl(url, filters)

  let exhaustedScrollers = 0
  let backwardWindow = [now() - interval, now()]
@@ -237,38 +172,26 @@ export const makeCalendarFeed = ({
    })
  }

-  const removeEvents = (ids: string[]) => {
-    events.update($events => $events.filter(e => !ids.includes(e.id)))
-  }
-
-  const onThunk = (thunk: Thunk) => {
-    if (matchFilters(feedFilters, thunk.event)) {
-      insertEvent(thunk.event)
-
-      thunk.controller.signal.addEventListener("abort", () => {
-        removeEvents([thunk.event.id])
-      })
+  const unsubscribe = on(repository, "update", ({added, removed}) => {
+    if (removed.size > 0) {
+      events.update($events => $events.filter(e => !removed.has(e.id)))
    }
-  }

-  request({
-    relays,
-    signal: controller.signal,
-    filters: subscriptionFilters,
-    onEvent: (e: TrustedEvent) => {
-      if (matchFilters(feedFilters, e)) insertEvent(e)
-    },
+    for (const event of added) {
+      if (matchFilters(filters, event)) {
+        insertEvent(event)
+      }
+    }
  })

  const loadTimeframe = (since: number, until: number) => {
    const hashes = daysBetween(since, until).map(String)

    request({
-      relays,
-      signal: controller.signal,
+      relays: [url],
      autoClose: true,
+      signal: controller.signal,
      filters: [{kinds: [EVENT_TIME], "#D": hashes}],
-      onEvent: insertEvent,
    })
  }

@@ -311,8 +234,6 @@ export const makeCalendarFeed = ({
    },
  })

-  const unsubscribe = thunkQueue.subscribe(onThunk)
-
  return {
    events,
    cleanup: () => {
@@ -5,6 +5,7 @@ import * as nip19 from "nostr-tools/nip19"
 import {
  on,
  call,
+  assoc,
  remove,
  uniqBy,
  sortBy,
@@ -38,6 +39,7 @@ import {isKindFeed, findFeed} from "@welshman/feeds"
 import {
  getIdFilters,
  WRAP,
+  DELETE,
  CLIENT_AUTH,
  AUTH_JOIN,
  REACTION,
@@ -50,6 +52,7 @@ import {
  ROOMS,
  THREAD,
  COMMENT,
+  REPORT,
  ROOM_JOIN,
  ROOM_ADD_USER,
  ROOM_REMOVE_USER,
@@ -60,6 +63,8 @@ import {
  ALERT_ANDROID,
  ALERT_STATUS,
  APP_DATA,
+  ZAP_GOAL,
+  EVENT_TIME,
  getGroupTags,
  getRelayTagValues,
  getPubkeyTagValues,
@@ -114,8 +119,6 @@ export const PROTECTED = ["-"]

 export const ENABLE_ZAPS = Capacitor.getPlatform() != "ios"

-export const REACTION_KINDS = ENABLE_ZAPS ? [REACTION, ZAP_RESPONSE] : [REACTION]
-
 export const NOTIFIER_PUBKEY = import.meta.env.VITE_NOTIFIER_PUBKEY

 export const NOTIFIER_RELAY = import.meta.env.VITE_NOTIFIER_RELAY
@@ -280,22 +283,27 @@ export const getUrlsForEvent = derived([trackerStore, thunks], ([$tracker, $thun
 })

 export const getEventsForUrl = (url: string, filters: Filter[]) => {
-  const $getUrlsForEvent = get(getUrlsForEvent)
-  const $events = repository.query(filters)
+  const ids = uniq([
+    ...tracker.getIds(url),
+    ...Array.from(flattenThunks(Object.values(get(thunks))))
+      .filter(t => t.options.relays.includes(url))
+      .map(t => t.event.id),
+  ])

-  return sortBy(
-    e => -e.created_at,
-    $events.filter(e => $getUrlsForEvent(e.id).includes(url)),
-  )
+  return repository.query(filters.map(assoc("ids", ids)))
 }

 export const deriveEventsForUrl = (url: string, filters: Filter[]) =>
-  derived([deriveEvents(repository, {filters}), getUrlsForEvent], ([$events, $getUrlsForEvent]) =>
-    sortBy(
-      e => -e.created_at,
-      $events.filter(e => $getUrlsForEvent(e.id).includes(url)),
-    ),
-  )
+  derived([trackerStore, thunks], ([$tracker, $thunks]) => {
+    const ids = uniq([
+      ...$tracker.getIds(url),
+      ...Array.from(flattenThunks(Object.values($thunks)))
+        .filter(t => t.options.relays.includes(url))
+        .map(t => t.event.id),
+    ])
+
+    return repository.query(filters.map(assoc("ids", ids)))
+  })

 // Context

@@ -306,6 +314,26 @@ routerContext.getIndexerRelays = always(INDEXER_RELAYS)
 netContext.isEventValid = (event: TrustedEvent, url: string) =>
  getSetting<string[]>("trusted_relays").includes(url) || verifyEvent(event)

+// Filters
+
+export const makeCommentFilter = (kinds: number[], extra: Filter = {}) => ({
+  kinds: [COMMENT],
+  "#K": kinds.map(String),
+  ...extra,
+})
+
+export const REACTION_KINDS = [REPORT, DELETE, REACTION]
+
+if (ENABLE_ZAPS) {
+  REACTION_KINDS.push(ZAP_RESPONSE)
+}
+
+export const MESSAGE_KINDS = [ZAP_GOAL, EVENT_TIME, THREAD, MESSAGE]
+
+export const MESSAGE_FILTER = {kinds: MESSAGE_KINDS}
+
+export const COMMENT_FILTER = makeCommentFilter(MESSAGE_KINDS)
+
 // Settings

 export const canDecrypt = synced({
@@ -1,19 +1,31 @@
 import {page} from "$app/stores"
 import type {Unsubscriber} from "svelte/store"
 import {derived, get} from "svelte/store"
-import {call, chunk, sleep, now, identity, WEEK, ago} from "@welshman/lib"
+import {
+  partition,
+  call,
+  sortBy,
+  assoc,
+  chunk,
+  sleep,
+  now,
+  identity,
+  WEEK,
+  MONTH,
+  ago,
+} from "@welshman/lib"
 import {
  getListTags,
  getRelayTagValues,
  WRAP,
-  MESSAGE,
-  ZAP_GOAL,
-  THREAD,
-  EVENT_TIME,
-  COMMENT,
+  ROOM_META,
+  ROOM_ADD_USER,
+  ROOM_REMOVE_USER,
  isSignedEvent,
+  normalizeRelayUrl,
 } from "@welshman/util"
-import {request, pull} from "@welshman/net"
+import type {Filter, TrustedEvent} from "@welshman/util"
+import {request, load, pull} from "@welshman/net"
 import {
  pubkey,
  loadRelay,
@@ -27,17 +39,54 @@ import {
  loadMutes,
  loadProfile,
  repository,
+  hasNegentropy,
 } from "@welshman/app"
 import {
+  MESSAGE_FILTER,
+  COMMENT_FILTER,
  INDEXER_RELAYS,
+  REACTION_KINDS,
  canDecrypt,
  loadSettings,
  userMembership,
  defaultPubkeys,
  decodeRelay,
  loadMembership,
+  getUrlsForEvent,
 } from "@app/core/state"
 import {loadAlerts, loadAlertStatuses} from "@app/core/requests"
+import {hasBlossomSupport} from "@app/core/commands"
+
+// Utils
+
+type PullOpts = {
+  relays: string[]
+  filters: Filter[]
+  signal: AbortSignal
+}
+
+const pullConservatively = ({relays, filters, signal}: PullOpts) => {
+  const $getUrlsForEvent = get(getUrlsForEvent)
+  const [smart, dumb] = partition(hasNegentropy, relays)
+  const events = repository.query(filters, {shouldSort: false}).filter(isSignedEvent)
+  const promises: Promise<TrustedEvent[]>[] = [pull({relays: smart, filters, signal, events})]
+
+  // Since pulling from relays without negentropy is expensive, limit how many
+  // duplicates we repeatedly download
+  for (const url of dumb) {
+    const urlEvents = events.filter(e => $getUrlsForEvent(e.id).includes(url))
+
+    if (urlEvents.length >= 100) {
+      filters = filters.map(assoc("since", sortBy(e => -e.created_at, urlEvents)[10]!.created_at))
+    }
+
+    promises.push(load({relays: [url], filters, signal}))
+  }
+
+  return Promise.all(promises)
+}
+
+// Relays

 const syncRelays = () => {
  for (const url of INDEXER_RELAYS) {
@@ -46,7 +95,10 @@ const syncRelays = () => {

  const unsubscribePage = page.subscribe($page => {
    if ($page.params.relay) {
-      loadRelay(decodeRelay($page.params.relay))
+      const url = decodeRelay($page.params.relay)
+
+      loadRelay(url)
+      hasBlossomSupport(url)
    }
  })

@@ -62,6 +114,8 @@ const syncRelays = () => {
  }
 }

+// User data
+
 const syncUserData = () => {
  const unsubscribePubkey = pubkey.subscribe($pubkey => {
    if ($pubkey) {
@@ -107,40 +161,40 @@ const syncUserData = () => {
  }
 }

-const syncSpace = (url: string) => {
+// Memberships
+
+const syncMembership = (url: string) => {
  const controller = new AbortController()

-  // Load historical data
-  pull({
+  // Load group metadata
+  pullConservatively({
    relays: [url],
    signal: controller.signal,
-    filters: [{kinds: [ZAP_GOAL, EVENT_TIME, THREAD, MESSAGE, COMMENT]}],
-    events: repository
-      .query([{kinds: [ZAP_GOAL, EVENT_TIME, THREAD, MESSAGE, COMMENT]}])
-      .filter(isSignedEvent),
+    filters: [{kinds: [ROOM_META]}],
  })

-  // Load new events
+  // Load historical data from up to a month ago for quick page loading
+  pullConservatively({
+    relays: [url],
+    signal: controller.signal,
+    filters: [MESSAGE_FILTER, COMMENT_FILTER].map(assoc("since", ago(MONTH))),
+  })
+
+  // Listen for new events
  request({
    relays: [url],
    signal: controller.signal,
-    filters: [{kinds: [ZAP_GOAL, EVENT_TIME, THREAD, MESSAGE, COMMENT], since: now()}],
+    filters: [MESSAGE_FILTER, COMMENT_FILTER].map(assoc("since", now())),
  })

  return () => controller.abort()
 }

-const syncSpaces = () => {
+const syncMemberships = () => {
  const unsubscribersByUrl = new Map<string, Unsubscriber>()
-  const unsubscribeMembership = userMembership.subscribe($l => {
-    const urls = getRelayTagValues(getListTags($l))

-    // Start syncing newly added spaces
-    for (const url of urls) {
-      if (!unsubscribersByUrl.has(url)) {
-        unsubscribersByUrl.set(url, syncSpace(url))
-      }
-    }
+  const unsubscribeMembership = userMembership.subscribe($l => {
+    const urls = getRelayTagValues(getListTags($l)).map(normalizeRelayUrl)

    // stop syncing removed spaces
    for (const [url, unsubscribe] of unsubscribersByUrl.entries()) {
@@ -149,6 +203,13 @@ const syncSpaces = () => {
        unsubscribe()
      }
    }
+
+    // Start syncing newly added spaces
+    for (const url of urls) {
+      if (!unsubscribersByUrl.has(url)) {
+        unsubscribersByUrl.set(url, syncMembership(url))
+      }
+    }
  })

  return () => {
@@ -157,17 +218,80 @@ const syncSpaces = () => {
  }
 }

+// Sync extra stuff for the current space
+
+const syncSpace = (url: string) => {
+  const $pubkey = pubkey.get()
+  const controller = new AbortController()
+
+  // Load all membership changes for the current user
+  if ($pubkey) {
+    pullConservatively({
+      relays: [url],
+      signal: controller.signal,
+      filters: [
+        {
+          kinds: [ROOM_ADD_USER, ROOM_REMOVE_USER],
+          "#p": [$pubkey],
+        },
+      ],
+    })
+  }
+
+  // Listen actively for all current membership changes, reports, reactions, zaps, etc
+  request({
+    relays: [url],
+    signal: controller.signal,
+    filters: [
+      {
+        kinds: [ROOM_ADD_USER, ROOM_REMOVE_USER, ...REACTION_KINDS],
+        since: now(),
+      },
+    ],
+  })
+
+  return () => controller.abort()
+}
+
+const syncCurrentSpace = () => {
+  const unsubscribersByUrl = new Map<string, Unsubscriber>()
+
+  // Sync the space the user is currently visiting
+  const unsubscribePage = page.subscribe($page => {
+    if ($page.params.relay) {
+      const url = decodeRelay($page.params.relay)
+
+      if (!unsubscribersByUrl.has(url)) {
+        unsubscribersByUrl.set(url, syncSpace(url))
+      }
+
+      for (const [oldUrl, unsubscribe] of unsubscribersByUrl.entries()) {
+        if (url !== oldUrl) {
+          unsubscribersByUrl.delete(oldUrl)
+          unsubscribe()
+        }
+      }
+    } else {
+      Array.from(unsubscribersByUrl.values()).forEach(call)
+    }
+  })
+
+  return () => {
+    Array.from(unsubscribersByUrl.values()).forEach(call)
+    unsubscribePage()
+  }
+}
+
+// DMs
+
 const syncDMRelay = (url: string, pubkey: string) => {
  const controller = new AbortController()

  // Load historical data
-  pull({
+  pullConservatively({
    relays: [url],
    signal: controller.signal,
    filters: [{kinds: [WRAP], "#p": [pubkey], until: ago(WEEK, 2)}],
-    events: repository
-      .query([{kinds: [ZAP_GOAL, EVENT_TIME, THREAD, MESSAGE, COMMENT]}])
-      .filter(isSignedEvent),
  })

  // Load new events
@@ -243,8 +367,16 @@ const syncDMs = () => {
  }
 }

+// Merge all synchronization functions
+
 export const syncApplicationData = () => {
-  const unsubscribers = [syncRelays(), syncUserData(), syncSpaces(), syncDMs()]
+  const unsubscribers = [
+    syncRelays(),
+    syncUserData(),
+    syncMemberships(),
+    syncCurrentSpace(),
+    syncDMs(),
+  ]

  return () => unsubscribers.forEach(call)
 }