Miscellaneous perf improvements and bsky sync-up (#1239)

* sync-up bsky appview to apply takedowns during hydration from #1232

* apply blocks and mutes on bsky appview custom algos

* hit index on with-friends algo pagination

* more performant query for fetching invite codes with uses

* tidy

* simplify query for actor typeahead for maximal use of trigram index

packages/bsky/src/api/app/bsky/actor/searchActorsTypeahead.ts

@@ -1,6 +1,9 @@
 import AppContext from '../../../../context'
 import { Server } from '../../../../lexicon'
-import { cleanTerm, getUserSearchQuery } from '../../../../services/util/search'
+import {
+  cleanTerm,
+  getUserSearchQuerySimple,
+} from '../../../../services/util/search'
 
 export default function (server: Server, ctx: AppContext) {
   server.app.bsky.actor.searchActorsTypeahead({
@@ -12,7 +15,7 @@ export default function (server: Server, ctx: AppContext) {
       const term = cleanTerm(rawTerm || '')
 
       const results = term
-        ? await getUserSearchQuery(db, { term, limit })
+        ? await getUserSearchQuerySimple(db, { term, limit })
             .selectAll('actor')
             .execute()
         : []

packages/bsky/src/api/app/bsky/feed/getPostThread.ts

@@ -127,9 +127,19 @@ const composeThread = (
 
   let replies: (ThreadViewPost | NotFoundPost | BlockedPost)[] | undefined
   if (threadData.replies) {
-    replies = threadData.replies.map((reply) =>
-      composeThread(reply, feedService, posts, actors, embeds, labels),
-    )
+    replies = threadData.replies.flatMap((reply) => {
+      const thread = composeThread(
+        reply,
+        feedService,
+        posts,
+        actors,
+        embeds,
+        labels,
+      )
+      // e.g. don't bother including #postNotFound reply placeholders for takedowns. either way matches api contract.
+      const skip = []
+      return isNotFoundPost(thread) ? skip : thread
+    })
   }
 
   return {

packages/bsky/src/api/app/bsky/util/feed.ts

@@ -12,7 +12,7 @@ export class FeedKeyset extends TimeCidKeyset<FeedRow> {
 }
 
 // For users with sparse feeds, avoid scanning more than one week for a single page
-export const getFeedDateThreshold = (from: string | undefined, days = 7) => {
+export const getFeedDateThreshold = (from: string | undefined, days = 3) => {
   const timelineDateThreshold = from ? new Date(from) : new Date()
   timelineDateThreshold.setDate(timelineDateThreshold.getDate() - days)
   return timelineDateThreshold.toISOString()

packages/bsky/src/db/pagination.ts

@@ -1,6 +1,6 @@
-import { SelectQueryBuilder, sql } from 'kysely'
+import { sql } from 'kysely'
 import { InvalidRequestError } from '@atproto/xrpc-server'
-import { DbRef } from './util'
+import { AnyQb, DbRef } from './util'
 
 export type Cursor = { primary: string; secondary: string }
 export type LabeledResult = {
@@ -107,7 +107,7 @@ export class TimeCidKeyset<
 }
 
 export const paginate = <
-  QB extends SelectQueryBuilder<any, any, any>,
+  QB extends AnyQb,
   K extends GenericKeyset<unknown, any>,
 >(
   qb: QB,

packages/bsky/src/db/util.ts

@@ -2,6 +2,7 @@ import {
   DummyDriver,
   DynamicModule,
   RawBuilder,
+  SelectQueryBuilder,
   sql,
   SqliteAdapter,
   SqliteIntrospector,
@@ -56,3 +57,5 @@ export const dummyDialect = {
 }
 
 export type DbRef = RawBuilder | ReturnType<DynamicModule['ref']>
+
+export type AnyQb = SelectQueryBuilder<any, any, any>

packages/bsky/src/feed-gen/best-of-follows.ts

@@ -7,30 +7,33 @@ import AppContext from '../context'
 const handler: AlgoHandler = async (
   ctx: AppContext,
   params: SkeletonParams,
-  requester: string,
+  viewer: string,
 ): Promise<AlgoResponse> => {
   const { limit, cursor } = params
   const feedService = ctx.services.feed(ctx.db)
+  const graphService = ctx.services.graph(ctx.db)
 
   const { ref } = ctx.db.db.dynamic
 
   // candidates are ranked within a materialized view by like count, depreciated over time.
 
-  // @TODO apply blocks and mutes
   let builder = feedService
     .selectPostQb()
     .innerJoin('algo_whats_hot_view as candidate', 'candidate.uri', 'post.uri')
-    .leftJoin('post_embed_record', 'post_embed_record.postUri', 'post.uri')
     .where((qb) =>
       qb
-        .where('post.creator', '=', requester)
+        .where('post.creator', '=', viewer)
         .orWhereExists((inner) =>
           inner
             .selectFrom('follow')
-            .where('follow.creator', '=', requester)
+            .where('follow.creator', '=', viewer)
             .whereRef('follow.subjectDid', '=', 'post.creator'),
         ),
     )
+    .where((qb) =>
+      graphService.whereNotMuted(qb, viewer, [ref('post.creator')]),
+    )
+    .whereNotExists(graphService.blockQb(viewer, [ref('post.creator')]))
     .select('candidate.score')
     .select('candidate.cid')
 

packages/bsky/src/feed-gen/bsky-team.ts

@@ -27,13 +27,17 @@ const handler: AlgoHandler = async (
 ): Promise<AlgoResponse> => {
   const { limit = 50, cursor } = params
   const feedService = ctx.services.feed(ctx.db)
+  const graphService = ctx.services.graph(ctx.db)
 
   const { ref } = ctx.db.db.dynamic
 
-  // @TODO apply blocks and mutes
   const postsQb = feedService
     .selectPostQb()
     .where('post.creator', 'in', BSKY_TEAM)
+    .where((qb) =>
+      graphService.whereNotMuted(qb, viewer, [ref('post.creator')]),
+    )
+    .whereNotExists(graphService.blockQb(viewer, [ref('post.creator')]))
 
   const keyset = new FeedKeyset(ref('sortAt'), ref('cid'))
 

packages/bsky/src/feed-gen/hot-classic.ts

@@ -23,10 +23,10 @@ const handler: AlgoHandler = async (
 ): Promise<AlgoResponse> => {
   const { limit = 50, cursor } = params
   const feedService = ctx.services.feed(ctx.db)
+  const graphService = ctx.services.graph(ctx.db)
 
   const { ref } = ctx.db.db.dynamic
 
-  // @TODO apply blocks and mutes
   const postsQb = feedService
     .selectPostQb()
     .leftJoin('post_agg', 'post_agg.uri', 'post.uri')
@@ -46,6 +46,10 @@ const handler: AlgoHandler = async (
             .orWhereRef('label.uri', '=', ref('post_embed_record.embedUri')),
         ),
     )
+    .where((qb) =>
+      graphService.whereNotMuted(qb, viewer, [ref('post.creator')]),
+    )
+    .whereNotExists(graphService.blockQb(viewer, [ref('post.creator')]))
 
   const keyset = new FeedKeyset(ref('sortAt'), ref('cid'))
 

packages/bsky/src/feed-gen/mutuals.ts

@@ -7,38 +7,42 @@ import { FeedKeyset, getFeedDateThreshold } from '../api/app/bsky/util/feed'
 const handler: AlgoHandler = async (
   ctx: AppContext,
   params: SkeletonParams,
-  requester: string,
+  viewer: string,
 ): Promise<AlgoResponse> => {
   const { limit = 50, cursor } = params
   const feedService = ctx.services.feed(ctx.db)
+  const graphService = ctx.services.graph(ctx.db)
 
   const { ref } = ctx.db.db.dynamic
 
   const mutualsSubquery = ctx.db.db
     .selectFrom('follow')
-    .where('follow.creator', '=', requester)
+    .where('follow.creator', '=', viewer)
     .whereExists((qb) =>
       qb
         .selectFrom('follow as follow_inner')
         .whereRef('follow_inner.creator', '=', 'follow.subjectDid')
-        .where('follow_inner.subjectDid', '=', requester)
+        .where('follow_inner.subjectDid', '=', viewer)
         .selectAll(),
     )
     .select('follow.subjectDid')
 
   const keyset = new FeedKeyset(ref('feed_item.sortAt'), ref('feed_item.cid'))
   const sortFrom = keyset.unpack(cursor)?.primary
 
-  // @TODO apply blocks and mutes
   let feedQb = feedService
     .selectFeedItemQb()
     .where('feed_item.type', '=', 'post') // ensures originatorDid is post.creator
     .where((qb) =>
       qb
-        .where('originatorDid', '=', requester)
+        .where('originatorDid', '=', viewer)
         .orWhere('originatorDid', 'in', mutualsSubquery),
     )
     .where('feed_item.sortAt', '>', getFeedDateThreshold(sortFrom))
+    .where((qb) =>
+      graphService.whereNotMuted(qb, viewer, [ref('originatorDid')]),
+    )
+    .whereNotExists(graphService.blockQb(viewer, [ref('originatorDid')]))
 
   feedQb = paginate(feedQb, { limit, cursor, keyset })
 

packages/bsky/src/feed-gen/whats-hot.ts

@@ -4,7 +4,7 @@ import { QueryParams as SkeletonParams } from '../lexicon/types/app/bsky/feed/ge
 import { AlgoHandler, AlgoResponse } from './types'
 import { GenericKeyset, paginate } from '../db/pagination'
 import AppContext from '../context'
-import { notSoftDeletedClause, valuesList } from '../db/util'
+import { valuesList } from '../db/util'
 import { sql } from 'kysely'
 import { FeedItemType } from '../services/types'
 
@@ -24,20 +24,16 @@ const handler: AlgoHandler = async (
   viewer: string,
 ): Promise<AlgoResponse> => {
   const { limit, cursor } = params
+  const graphService = ctx.services.graph(ctx.db)
 
   const { ref } = ctx.db.db.dynamic
 
   // candidates are ranked within a materialized view by like count, depreciated over time.
 
-  // @TODO apply blocks and mutes
   let builder = ctx.db.db
     .selectFrom('algo_whats_hot_view as candidate')
     .innerJoin('post', 'post.uri', 'candidate.uri')
-    .innerJoin('actor as author', 'author.did', 'post.creator')
-    .innerJoin('record', 'record.uri', 'post.uri')
     .leftJoin('post_embed_record', 'post_embed_record.postUri', 'candidate.uri')
-    .where(notSoftDeletedClause(ref('author')))
-    .where(notSoftDeletedClause(ref('record')))
     .whereNotExists((qb) =>
       qb
         .selectFrom('label')
@@ -51,6 +47,10 @@ const handler: AlgoHandler = async (
             .orWhereRef('label.uri', '=', ref('post_embed_record.embedUri')),
         ),
     )
+    .where((qb) =>
+      graphService.whereNotMuted(qb, viewer, [ref('post.creator')]),
+    )
+    .whereNotExists(graphService.blockQb(viewer, [ref('post.creator')]))
     .select([
       sql<FeedItemType>`${'post'}`.as('type'),
       'post.uri as uri',

packages/bsky/src/feed-gen/with-friends.ts

@@ -2,7 +2,7 @@ import AppContext from '../context'
 import { QueryParams as SkeletonParams } from '../lexicon/types/app/bsky/feed/getFeedSkeleton'
 import { paginate } from '../db/pagination'
 import { AlgoHandler, AlgoResponse } from './types'
-import { FeedKeyset } from '../api/app/bsky/util/feed'
+import { FeedKeyset, getFeedDateThreshold } from '../api/app/bsky/util/feed'
 
 const handler: AlgoHandler = async (
   ctx: AppContext,
@@ -11,26 +11,30 @@ const handler: AlgoHandler = async (
 ): Promise<AlgoResponse> => {
   const { cursor, limit = 50 } = params
   const feedService = ctx.services.feed(ctx.db)
+  const graphService = ctx.services.graph(ctx.db)
 
   const { ref } = ctx.db.db.dynamic
 
-  // @NOTE use of getFeedDateThreshold() not currently beneficial to this feed
-  const keyset = new FeedKeyset(ref('feed_item.sortAt'), ref('feed_item.cid'))
+  const keyset = new FeedKeyset(ref('post.indexedAt'), ref('post.cid'))
+  const sortFrom = keyset.unpack(cursor)?.primary
 
-  // @TODO apply blocks and mutes
   let postsQb = feedService
-    .selectFeedItemQb()
-    .innerJoin('post_agg', 'post_agg.uri', 'feed_item.uri')
-    .where('feed_item.type', '=', 'post')
+    .selectPostQb()
+    .innerJoin('post_agg', 'post_agg.uri', 'post.uri')
     .where('post_agg.likeCount', '>=', 5)
     .whereExists((qb) =>
       qb
         .selectFrom('follow')
         .where('follow.creator', '=', requester)
-        .whereRef('follow.subjectDid', '=', 'originatorDid'),
+        .whereRef('follow.subjectDid', '=', 'post.creator'),
     )
+    .where((qb) =>
+      graphService.whereNotMuted(qb, requester, [ref('post.creator')]),
+    )
+    .whereNotExists(graphService.blockQb(requester, [ref('post.creator')]))
+    .where('post.indexedAt', '>', getFeedDateThreshold(sortFrom))
 
-  postsQb = paginate(postsQb, { limit, cursor, keyset })
+  postsQb = paginate(postsQb, { limit, cursor, keyset, tryIndex: true })
 
   const feedItems = await postsQb.execute()
   return {
@@ -40,51 +44,3 @@ const handler: AlgoHandler = async (
 }
 
 export default handler
-
-// Original algorithm, temporarily disabled because of performance issues
-// --------------------------
-
-// const postRate = sql`(10000 * ${ref('postsCount')} / extract(epoch from ${ref(
-//   'user_account.createdAt',
-// )}::timestamp))`
-// const mostActiveMutuals = await ctx.db.db
-//   .selectFrom('follow')
-//   .select('subjectDid as did')
-//   .innerJoin('user_account', 'user_account.did', 'follow.subjectDid')
-//   .innerJoin('profile_agg', 'profile_agg.did', 'follow.subjectDid')
-//   .where('follow.creator', '=', requester)
-//   .whereExists((qb) =>
-//     qb
-//       .selectFrom('follow as mutual')
-//       .where('mutual.subjectDid', '=', requester)
-//       .whereRef('mutual.creator', '=', 'follow.subjectDid'),
-//   )
-//   .orderBy(postRate, 'desc')
-//   .limit(25)
-//   .execute()
-
-// if (!mostActiveMutuals.length) {
-//   return { feedItems: [] }
-// }
-
-// // All posts that hit a certain threshold of likes and also have
-// // at least one like by one of your most active mutuals.
-// let postsQb = feedService
-//   .selectFeedItemQb()
-//   .innerJoin('post_agg', 'post_agg.uri', 'feed_item.uri')
-//   .where('feed_item.type', '=', 'post')
-//   .where('post_agg.likeCount', '>=', 5)
-//   .whereExists((qb) => {
-//     return qb
-//       .selectFrom('like')
-//       .whereRef('like.subject', '=', 'post.uri')
-//       .whereRef(
-//         'like.creator',
-//         'in',
-//         valuesList(mostActiveMutuals.map((follow) => follow.did)),
-//       )
-//   })
-//   .where((qb) =>
-//     accountService.whereNotMuted(qb, requester, [ref('post.creator')]),
-//   )
-//   .whereNotExists(graphService.blockQb(requester, [ref('post.creator')]))