4 files changed, 557 insertions, 0 deletions
diff --git a/src/state/queries/notifications/feed.ts b/src/state/queries/notifications/feed.ts
new file mode 100644
index 000000000..16025f856
--- /dev/null
+++ b/src/state/queries/notifications/feed.ts
@@ -0,0 +1,125 @@
+/**
+ * NOTE
+ * The ./unread.ts API:
+ *
+ * - Provides a `checkUnread()` function to sync with the server,
+ * - Periodically calls `checkUnread()`, and
+ * - Caches the first page of notifications.
+ *
+ * IMPORTANT: This query uses ./unread.ts's cache as its first page,
+ * IMPORTANT: which means the cache-freshness of this query is driven by the unread API.
+ *
+ * Follow these rules:
+ *
+ * 1. Call `checkUnread()` if you want to fetch latest in the background.
+ * 2. Call `checkUnread({invalidate: true})` if you want latest to sync into this query's results immediately.
+ * 3. Don't call this query's `refetch()` if you're trying to sync latest; call `checkUnread()` instead.
+ */
+
+import {AppBskyFeedDefs} from '@atproto/api'
+import {
+  useInfiniteQuery,
+  InfiniteData,
+  QueryKey,
+  useQueryClient,
+  QueryClient,
+} from '@tanstack/react-query'
+import {useModerationOpts} from '../preferences'
+import {useUnreadNotificationsApi} from './unread'
+import {fetchPage} from './util'
+import {FeedPage} from './types'
+import {useMutedThreads} from '#/state/muted-threads'
+import {STALE} from '..'
+
+export type {NotificationType, FeedNotification, FeedPage} from './types'
+
+const PAGE_SIZE = 30
+
+type RQPageParam = string | undefined
+
+export function RQKEY() {
+  return ['notification-feed']
+}
+
+export function useNotificationFeedQuery(opts?: {enabled?: boolean}) {
+  const queryClient = useQueryClient()
+  const moderationOpts = useModerationOpts()
+  const threadMutes = useMutedThreads()
+  const unreads = useUnreadNotificationsApi()
+  const enabled = opts?.enabled !== false
+
+  return useInfiniteQuery<
+    FeedPage,
+    Error,
+    InfiniteData<FeedPage>,
+    QueryKey,
+    RQPageParam
+  >({
+    staleTime: STALE.INFINITY,
+    queryKey: RQKEY(),
+    async queryFn({pageParam}: {pageParam: RQPageParam}) {
+      let page
+      if (!pageParam) {
+        // for the first page, we check the cached page held by the unread-checker first
+        page = unreads.getCachedUnreadPage()
+      }
+      if (!page) {
+        page = await fetchPage({
+          limit: PAGE_SIZE,
+          cursor: pageParam,
+          queryClient,
+          moderationOpts,
+          threadMutes,
+        })
+      }
+
+      // if the first page has an unread, mark all read
+      if (!pageParam && page.items[0] && !page.items[0].notification.isRead) {
+        unreads.markAllRead()
+      }
+
+      return page
+    },
+    initialPageParam: undefined,
+    getNextPageParam: lastPage => lastPage.cursor,
+    enabled,
+  })
+}
+
+/**
+ * This helper is used by the post-thread placeholder function to
+ * find a post in the query-data cache
+ */
+export function findPostInQueryData(
+  queryClient: QueryClient,
+  uri: string,
+): AppBskyFeedDefs.PostView | undefined {
+  const generator = findAllPostsInQueryData(queryClient, uri)
+  const result = generator.next()
+  if (result.done) {
+    return undefined
+  } else {
+    return result.value
+  }
+}
+
+export function* findAllPostsInQueryData(
+  queryClient: QueryClient,
+  uri: string,
+): Generator<AppBskyFeedDefs.PostView, void> {
+  const queryDatas = queryClient.getQueriesData<InfiniteData<FeedPage>>({
+    queryKey: ['notification-feed'],
+  })
+  for (const [_queryKey, queryData] of queryDatas) {
+    if (!queryData?.pages) {
+      continue
+    }
+    for (const page of queryData?.pages) {
+      for (const item of page.items) {
+        if (item.subject?.uri === uri) {
+          yield item.subject
+        }
+      }
+    }
+  }
+}
diff --git a/src/state/queries/notifications/types.ts b/src/state/queries/notifications/types.ts
new file mode 100644
index 000000000..0e88f1071
--- /dev/null
+++ b/src/state/queries/notifications/types.ts
@@ -0,0 +1,34 @@
+import {
+  AppBskyNotificationListNotifications,
+  AppBskyFeedDefs,
+} from '@atproto/api'
+
+export type NotificationType =
+  | 'post-like'
+  | 'feedgen-like'
+  | 'repost'
+  | 'mention'
+  | 'reply'
+  | 'quote'
+  | 'follow'
+  | 'unknown'
+
+export interface FeedNotification {
+  _reactKey: string
+  type: NotificationType
+  notification: AppBskyNotificationListNotifications.Notification
+  additional?: AppBskyNotificationListNotifications.Notification[]
+  subjectUri?: string
+  subject?: AppBskyFeedDefs.PostView
+}
+
+export interface FeedPage {
+  cursor: string | undefined
+  items: FeedNotification[]
+}
+
+export interface CachedFeedPage {
+  sessDid: string // used to invalidate on session changes
+  syncedAt: Date
+  data: FeedPage | undefined
+}
diff --git a/src/state/queries/notifications/unread.tsx b/src/state/queries/notifications/unread.tsx
new file mode 100644
index 000000000..6c130aaea
--- /dev/null
+++ b/src/state/queries/notifications/unread.tsx
@@ -0,0 +1,179 @@
+/**
+ * A kind of companion API to ./feed.ts. See that file for more info.
+ */
+
+import React from 'react'
+import * as Notifications from 'expo-notifications'
+import {useQueryClient} from '@tanstack/react-query'
+import BroadcastChannel from '#/lib/broadcast'
+import {useSession, getAgent} from '#/state/session'
+import {useModerationOpts} from '../preferences'
+import {fetchPage} from './util'
+import {CachedFeedPage, FeedPage} from './types'
+import {isNative} from '#/platform/detection'
+import {useMutedThreads} from '#/state/muted-threads'
+import {RQKEY as RQKEY_NOTIFS} from './feed'
+import {logger} from '#/logger'
+import {truncateAndInvalidate} from '../util'
+
+const UPDATE_INTERVAL = 30 * 1e3 // 30sec
+
+const broadcast = new BroadcastChannel('NOTIFS_BROADCAST_CHANNEL')
+
+type StateContext = string
+
+interface ApiContext {
+  markAllRead: () => Promise<void>
+  checkUnread: (opts?: {invalidate?: boolean}) => Promise<void>
+  getCachedUnreadPage: () => FeedPage | undefined
+}
+
+const stateContext = React.createContext<StateContext>('')
+
+const apiContext = React.createContext<ApiContext>({
+  async markAllRead() {},
+  async checkUnread() {},
+  getCachedUnreadPage: () => undefined,
+})
+
+export function Provider({children}: React.PropsWithChildren<{}>) {
+  const {hasSession, currentAccount} = useSession()
+  const queryClient = useQueryClient()
+  const moderationOpts = useModerationOpts()
+  const threadMutes = useMutedThreads()
+
+  const [numUnread, setNumUnread] = React.useState('')
+
+  const checkUnreadRef = React.useRef<ApiContext['checkUnread'] | null>(null)
+  const cacheRef = React.useRef<CachedFeedPage>({
+    sessDid: currentAccount?.did || '',
+    syncedAt: new Date(),
+    data: undefined,
+  })
+
+  // periodic sync
+  React.useEffect(() => {
+    if (!hasSession || !checkUnreadRef.current) {
+      return
+    }
+    checkUnreadRef.current() // fire on init
+    const interval = setInterval(checkUnreadRef.current, UPDATE_INTERVAL)
+    return () => clearInterval(interval)
+  }, [hasSession])
+
+  // listen for broadcasts
+  React.useEffect(() => {
+    const listener = ({data}: MessageEvent) => {
+      cacheRef.current = {
+        sessDid: currentAccount?.did || '',
+        syncedAt: new Date(),
+        data: undefined,
+      }
+      setNumUnread(data.event)
+    }
+    broadcast.addEventListener('message', listener)
+    return () => {
+      broadcast.removeEventListener('message', listener)
+    }
+  }, [setNumUnread, currentAccount])
+
+  // create API
+  const api = React.useMemo<ApiContext>(() => {
+    return {
+      async markAllRead() {
+        // update server
+        await getAgent().updateSeenNotifications(
+          cacheRef.current.syncedAt.toISOString(),
+        )
+
+        // update & broadcast
+        setNumUnread('')
+        broadcast.postMessage({event: ''})
+      },
+
+      async checkUnread({invalidate}: {invalidate?: boolean} = {}) {
+        try {
+          if (!getAgent().session) return
+
+          // count
+          const page = await fetchPage({
+            cursor: undefined,
+            limit: 40,
+            queryClient,
+            moderationOpts,
+            threadMutes,
+          })
+          const unreadCount = countUnread(page)
+          const unreadCountStr =
+            unreadCount >= 30
+              ? '30+'
+              : unreadCount === 0
+              ? ''
+              : String(unreadCount)
+          if (isNative) {
+            Notifications.setBadgeCountAsync(Math.min(unreadCount, 30))
+          }
+
+          // track last sync
+          const now = new Date()
+          const lastIndexed =
+            page.items[0] && new Date(page.items[0].notification.indexedAt)
+          cacheRef.current = {
+            sessDid: currentAccount?.did || '',
+            data: page,
+            syncedAt: !lastIndexed || now > lastIndexed ? now : lastIndexed,
+          }
+
+          // update & broadcast
+          setNumUnread(unreadCountStr)
+          if (invalidate) {
+            truncateAndInvalidate(queryClient, RQKEY_NOTIFS())
+          }
+          broadcast.postMessage({event: unreadCountStr})
+        } catch (e) {
+          logger.error('Failed to check unread notifications', {error: e})
+        }
+      },
+
+      getCachedUnreadPage() {
+        // return cached page if was for the current user
+        // (protects against session changes serving data from the past session)
+        if (cacheRef.current.sessDid === currentAccount?.did) {
+          return cacheRef.current.data
+        }
+      },
+    }
+  }, [setNumUnread, queryClient, moderationOpts, threadMutes, currentAccount])
+  checkUnreadRef.current = api.checkUnread
+
+  return (
+    <stateContext.Provider value={numUnread}>
+      <apiContext.Provider value={api}>{children}</apiContext.Provider>
+    </stateContext.Provider>
+  )
+}
+
+export function useUnreadNotifications() {
+  return React.useContext(stateContext)
+}
+
+export function useUnreadNotificationsApi() {
+  return React.useContext(apiContext)
+}
+
+function countUnread(page: FeedPage) {
+  let num = 0
+  for (const item of page.items) {
+    if (!item.notification.isRead) {
+      num++
+    }
+    if (item.additional) {
+      for (const item2 of item.additional) {
+        if (!item2.isRead) {
+          num++
+        }
+      }
+    }
+  }
+  return num
+}
diff --git a/src/state/queries/notifications/util.ts b/src/state/queries/notifications/util.ts
new file mode 100644
index 000000000..b8f320473
--- /dev/null
+++ b/src/state/queries/notifications/util.ts
@@ -0,0 +1,219 @@
+import {
+  AppBskyNotificationListNotifications,
+  ModerationOpts,
+  moderateProfile,
+  moderatePost,
+  AppBskyFeedDefs,
+  AppBskyFeedPost,
+  AppBskyFeedRepost,
+  AppBskyFeedLike,
+} from '@atproto/api'
+import chunk from 'lodash.chunk'
+import {QueryClient} from '@tanstack/react-query'
+import {getAgent} from '../../session'
+import {precacheProfile as precacheResolvedUri} from '../resolve-uri'
+import {NotificationType, FeedNotification, FeedPage} from './types'
+
+const GROUPABLE_REASONS = ['like', 'repost', 'follow']
+const MS_1HR = 1e3 * 60 * 60
+const MS_2DAY = MS_1HR * 48
+
+// exported api
+// =
+
+export async function fetchPage({
+  cursor,
+  limit,
+  queryClient,
+  moderationOpts,
+  threadMutes,
+}: {
+  cursor: string | undefined
+  limit: number
+  queryClient: QueryClient
+  moderationOpts: ModerationOpts | undefined
+  threadMutes: string[]
+}): Promise<FeedPage> {
+  const res = await getAgent().listNotifications({
+    limit,
+    cursor,
+  })
+
+  // filter out notifs by mod rules
+  const notifs = res.data.notifications.filter(
+    notif => !shouldFilterNotif(notif, moderationOpts),
+  )
+
+  // group notifications which are essentially similar (follows, likes on a post)
+  let notifsGrouped = groupNotifications(notifs)
+
+  // we fetch subjects of notifications (usually posts) now instead of lazily
+  // in the UI to avoid relayouts
+  const subjects = await fetchSubjects(notifsGrouped)
+  for (const notif of notifsGrouped) {
+    if (notif.subjectUri) {
+      notif.subject = subjects.get(notif.subjectUri)
+      if (notif.subject) {
+        precacheResolvedUri(queryClient, notif.subject.author) // precache the handle->did resolution
+      }
+    }
+  }
+
+  // apply thread muting
+  notifsGrouped = notifsGrouped.filter(
+    notif => !isThreadMuted(notif, threadMutes),
+  )
+
+  return {
+    cursor: res.data.cursor,
+    items: notifsGrouped,
+  }
+}
+
+// internal methods
+// =
+
+// TODO this should be in the sdk as moderateNotification -prf
+function shouldFilterNotif(
+  notif: AppBskyNotificationListNotifications.Notification,
+  moderationOpts: ModerationOpts | undefined,
+): boolean {
+  if (!moderationOpts) {
+    return false
+  }
+  const profile = moderateProfile(notif.author, moderationOpts)
+  if (
+    profile.account.filter ||
+    profile.profile.filter ||
+    notif.author.viewer?.muted
+  ) {
+    return true
+  }
+  if (
+    notif.type === 'reply' ||
+    notif.type === 'quote' ||
+    notif.type === 'mention'
+  ) {
+    // NOTE: the notification overlaps the post enough for this to work
+    const post = moderatePost(notif, moderationOpts)
+    if (post.content.filter) {
+      return true
+    }
+  }
+  // TODO: thread muting is not being applied
+  // (this requires fetching the post)
+  return false
+}
+
+function groupNotifications(
+  notifs: AppBskyNotificationListNotifications.Notification[],
+): FeedNotification[] {
+  const groupedNotifs: FeedNotification[] = []
+  for (const notif of notifs) {
+    const ts = +new Date(notif.indexedAt)
+    let grouped = false
+    if (GROUPABLE_REASONS.includes(notif.reason)) {
+      for (const groupedNotif of groupedNotifs) {
+        const ts2 = +new Date(groupedNotif.notification.indexedAt)
+        if (
+          Math.abs(ts2 - ts) < MS_2DAY &&
+          notif.reason === groupedNotif.notification.reason &&
+          notif.reasonSubject === groupedNotif.notification.reasonSubject &&
+          notif.author.did !== groupedNotif.notification.author.did &&
+          notif.isRead === groupedNotif.notification.isRead
+        ) {
+          groupedNotif.additional = groupedNotif.additional || []
+          groupedNotif.additional.push(notif)
+          grouped = true
+          break
+        }
+      }
+    }
+    if (!grouped) {
+      const type = toKnownType(notif)
+      groupedNotifs.push({
+        _reactKey: `notif-${notif.uri}`,
+        type,
+        notification: notif,
+        subjectUri: getSubjectUri(type, notif),
+      })
+    }
+  }
+  return groupedNotifs
+}
+
+async function fetchSubjects(
+  groupedNotifs: FeedNotification[],
+): Promise<Map<string, AppBskyFeedDefs.PostView>> {
+  const uris = new Set<string>()
+  for (const notif of groupedNotifs) {
+    if (notif.subjectUri) {
+      uris.add(notif.subjectUri)
+    }
+  }
+  const uriChunks = chunk(Array.from(uris), 25)
+  const postsChunks = await Promise.all(
+    uriChunks.map(uris =>
+      getAgent()
+        .app.bsky.feed.getPosts({uris})
+        .then(res => res.data.posts),
+    ),
+  )
+  const map = new Map<string, AppBskyFeedDefs.PostView>()
+  for (const post of postsChunks.flat()) {
+    if (
+      AppBskyFeedPost.isRecord(post.record) &&
+      AppBskyFeedPost.validateRecord(post.record).success
+    ) {
+      map.set(post.uri, post)
+    }
+  }
+  return map
+}
+
+function toKnownType(
+  notif: AppBskyNotificationListNotifications.Notification,
+): NotificationType {
+  if (notif.reason === 'like') {
+    if (notif.reasonSubject?.includes('feed.generator')) {
+      return 'feedgen-like'
+    }
+    return 'post-like'
+  }
+  if (
+    notif.reason === 'repost' ||
+    notif.reason === 'mention' ||
+    notif.reason === 'reply' ||
+    notif.reason === 'quote' ||
+    notif.reason === 'follow'
+  ) {
+    return notif.reason as NotificationType
+  }
+  return 'unknown'
+}
+
+function getSubjectUri(
+  type: NotificationType,
+  notif: AppBskyNotificationListNotifications.Notification,
+): string | undefined {
+  if (type === 'reply' || type === 'quote' || type === 'mention') {
+    return notif.uri
+  } else if (type === 'post-like' || type === 'repost') {
+    if (
+      AppBskyFeedRepost.isRecord(notif.record) ||
+      AppBskyFeedLike.isRecord(notif.record)
+    ) {
+      return typeof notif.record.subject?.uri === 'string'
+        ? notif.record.subject?.uri
+        : undefined
+    }
+  }
+}
+
+function isThreadMuted(notif: FeedNotification, mutes: string[]): boolean {
+  if (!notif.subject) {
+    return false
+  }
+  const record = notif.subject.record as AppBskyFeedPost.Record // assured in fetchSubjects()
+  return mutes.includes(record.reply?.root.uri || notif.subject.uri)
+}