Build timeline and author feeds from feed items index (bluesky-social…

…#774)

* Index posts and reposts into feed_item table for building feeds

* Use feed_item table for building timeline and author feeds

* Apply feed item indexing to bsky app view

* Fix bsky appview tests, test getPopular

* Use feed item index to build feeds in bsky app view

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

@@ -30,16 +30,15 @@ export default function (server: Server, ctx: AppContext) {
       }
 
       // @NOTE mutes applied on pds
-      const postsQb = feedService.selectPostQb().where('post.creator', '=', did)
+      let feedItemsQb = feedService
+        .selectFeedItemQb()
+        .where('originatorDid', '=', did)
 
-      const repostsQb = feedService
-        .selectRepostQb()
-        .where('repost.creator', '=', did)
+      const keyset = new FeedKeyset(
+        ref('feed_item.sortAt'),
+        ref('feed_item.cid'),
+      )
 
-      const keyset = new FeedKeyset(ref('cursor'), ref('postCid'))
-      let feedItemsQb = db
-        .selectFrom(postsQb.unionAll(repostsQb).as('feed_items'))
-        .selectAll()
       feedItemsQb = paginate(feedItemsQb, {
         limit,
         cursor,

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

@@ -116,7 +116,7 @@ const getRelevantIds = (
       fromChild.uris.forEach((uri) => uris.add(uri))
     }
   }
-  dids.add(thread.post.authorDid)
+  dids.add(thread.post.postAuthorDid)
   uris.add(thread.post.postUri)
   return { dids, uris }
 }

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

@@ -27,26 +27,19 @@ export default function (server: Server, ctx: AppContext) {
         .where('follow.creator', '=', requester)
 
       // @NOTE mutes applied on pds
-      const postsQb = feedService
-        .selectPostQb()
+      let feedItemsQb = feedService
+        .selectFeedItemQb()
         .where((qb) =>
           qb
-            .where('creator', '=', requester)
-            .orWhere('creator', 'in', followingIdsSubquery),
+            .where('originatorDid', '=', requester)
+            .orWhere('originatorDid', 'in', followingIdsSubquery),
         )
 
-      const repostsQb = feedService
-        .selectRepostQb()
-        .where((qb) =>
-          qb
-            .where('repost.creator', '=', requester)
-            .orWhere('repost.creator', 'in', followingIdsSubquery),
-        )
+      const keyset = new FeedKeyset(
+        ref('feed_item.sortAt'),
+        ref('feed_item.cid'),
+      )
 
-      const keyset = new FeedKeyset(ref('cursor'), ref('postCid'))
-      let feedItemsQb = db
-        .selectFrom(postsQb.unionAll(repostsQb).as('feed_items'))
-        .selectAll()
       feedItemsQb = paginate(feedItemsQb, {
         limit,
         cursor,

packages/bsky/src/api/app/bsky/unspecced.ts

@@ -5,6 +5,7 @@ import { paginate } from '../../../db/pagination'
 import AppContext from '../../../context'
 import { FeedRow, FeedItemType } from '../../../services/types'
 import { authOptionalVerifier } from './util'
+import { countAll } from '../../../db/util'
 
 // THIS IS A TEMPORARY UNSPECCED ROUTE
 export default function (server: Server, ctx: AppContext) {
@@ -19,34 +20,36 @@ export default function (server: Server, ctx: AppContext) {
       const feedService = ctx.services.feed(ctx.db)
 
       const postsQb = ctx.db.db
-        .with('like_counts', (qb) =>
-          qb
-            .selectFrom('like')
-            .groupBy('like.subject')
-            .select([sql`count(*)`.as('count'), 'like.subject']),
-        )
         .selectFrom('post')
-        .innerJoin('like_counts', 'like_counts.subject', 'post.uri')
         .leftJoin('repost', (join) =>
           // this works well for one curating user. reassess if adding more
           join
             .on('repost.creator', '=', 'did:plc:ea2eqamjmtuo6f4rvhl3g6ne')
             .onRef('repost.subject', '=', 'post.uri'),
         )
-        .where('like_counts.count', '>=', 5)
+        .where(
+          (qb) =>
+            qb
+              .selectFrom('like')
+              .whereRef('like.subject', '=', 'post.uri')
+              .select(countAll.as('count')),
+          '>=',
+          5,
+        )
         .orWhere('repost.creator', 'is not', null)
         .select([
           sql<FeedItemType>`${'post'}`.as('type'),
+          'post.uri as uri',
+          'post.cid as cid',
           'post.uri as postUri',
-          'post.cid as postCid',
+          'post.creator as postAuthorDid',
           'post.creator as originatorDid',
-          'post.creator as authorDid',
           'post.replyParent as replyParent',
           'post.replyRoot as replyRoot',
-          'post.indexedAt as cursor',
+          'post.indexedAt as sortAt',
         ])
 
-      const keyset = new FeedKeyset(ref('cursor'), ref('postCid'))
+      const keyset = new FeedKeyset(ref('sortAt'), ref('cid'))
 
       let feedQb = ctx.db.db.selectFrom(postsQb.as('feed_items')).selectAll()
       feedQb = paginate(feedQb, { limit, cursor, keyset })

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

@@ -15,7 +15,7 @@ export const composeFeed = async (
   const postUris = new Set<string>()
   for (const row of rows) {
     actorDids.add(row.originatorDid)
-    actorDids.add(row.authorDid)
+    actorDids.add(row.postAuthorDid)
     postUris.add(row.postUri)
     if (row.replyParent) {
       postUris.add(row.replyParent)
@@ -54,7 +54,7 @@ export const composeFeed = async (
           ? {
               $type: reasonType,
               by: actors[row.originatorDid],
-              indexedAt: row.cursor,
+              indexedAt: row.sortAt,
             }
           : undefined,
         reply:
@@ -76,6 +76,6 @@ export enum FeedAlgorithm {
 
 export class FeedKeyset extends TimeCidKeyset<FeedRow> {
   labelResult(result: FeedRow) {
-    return { primary: result.cursor, secondary: result.postCid }
+    return { primary: result.sortAt, secondary: result.cid }
   }
 }

packages/bsky/src/api/index.ts

@@ -12,6 +12,7 @@ import getFollows from './app/bsky/graph/getFollows'
 import searchActors from './app/bsky/actor/searchActors'
 import searchActorsTypeahead from './app/bsky/actor/searchActorsTypeahead'
 import getSuggestions from './app/bsky/actor/getSuggestions'
+import unspecced from './app/bsky/unspecced'
 
 export * as health from './health'
 
@@ -30,5 +31,6 @@ export default function (server: Server, ctx: AppContext) {
   searchActors(server, ctx)
   searchActorsTypeahead(server, ctx)
   getSuggestions(server, ctx)
+  unspecced(server, ctx)
   return server
 }

packages/bsky/src/services/feed/index.ts

@@ -27,36 +27,37 @@ export class FeedService {
       .where(notSoftDeletedClause(ref('record')))
       .select([
         sql<FeedItemType>`${'post'}`.as('type'),
+        'post.uri as uri',
+        'post.cid as cid',
         'post.uri as postUri',
-        'post.cid as postCid',
         'post.creator as originatorDid',
-        'post.creator as authorDid',
+        'post.creator as postAuthorDid',
         'post.replyParent as replyParent',
         'post.replyRoot as replyRoot',
-        'post.sortAt as cursor',
+        'post.sortAt as sortAt',
       ])
   }
 
-  selectRepostQb() {
+  selectFeedItemQb() {
     const { ref } = this.db.db.dynamic
     return this.db.db
-      .selectFrom('repost')
-      .innerJoin('post', 'post.uri', 'repost.subject')
+      .selectFrom('feed_item')
+      .innerJoin('post', 'post.uri', 'feed_item.postUri')
       .innerJoin('actor as author', 'author.did', 'post.creator')
-      .innerJoin('actor as originator', 'originator.did', 'repost.creator')
+      .innerJoin(
+        'actor as originator',
+        'originator.did',
+        'feed_item.originatorDid',
+      )
       .innerJoin('record as post_record', 'post_record.uri', 'post.uri')
       .where(notSoftDeletedClause(ref('author')))
       .where(notSoftDeletedClause(ref('originator')))
       .where(notSoftDeletedClause(ref('post_record')))
+      .selectAll('feed_item')
       .select([
-        sql<FeedItemType>`${'repost'}`.as('type'),
-        'post.uri as postUri',
-        'post.cid as postCid',
-        'repost.creator as originatorDid',
-        'post.creator as authorDid',
-        'post.replyParent as replyParent',
-        'post.replyRoot as replyRoot',
-        'repost.sortAt as cursor',
+        'post.replyRoot',
+        'post.replyParent',
+        'post.creator as postAuthorDid',
       ])
   }
 

packages/bsky/src/services/types.ts

@@ -36,11 +36,12 @@ export type FeedItemType = 'post' | 'repost'
 
 export type FeedRow = {
   type: FeedItemType
+  uri: string
+  cid: string
   postUri: string
-  postCid: string
+  postAuthorDid: string
   originatorDid: string
-  authorDid: string
   replyParent: string | null
   replyRoot: string | null
-  cursor: string
+  sortAt: string
 }

packages/bsky/tests/_util.ts

@@ -333,6 +333,7 @@ export const processAll = async (server: TestServerInfo, timeout = 5000) => {
   const start = Date.now()
   while (Date.now() - start < timeout) {
     await wait(50)
+    if (!sub) return
     const state = await sub.getState()
     const { lastSeq } = await db
       .selectFrom('repo_seq')

packages/bsky/tests/views/popular.test.ts

@@ -0,0 +1,93 @@
+import AtpAgent from '@atproto/api'
+import { runTestServer, CloseFn, processAll, TestServerInfo } from '../_util'
+import { SeedClient } from '../seeds/client'
+import basicSeed from '../seeds/basic'
+
+describe('popular views', () => {
+  let server: TestServerInfo
+  let agent: AtpAgent
+  let close: CloseFn
+  let sc: SeedClient
+
+  // account dids, for convenience
+  let alice: string
+  let bob: string
+  let carol: string
+  let dan: string
+  let eve: string
+  let frank: string
+
+  const account = {
+    email: '[email protected]',
+    password: 'blh-pass',
+  }
+
+  beforeAll(async () => {
+    server = await runTestServer({
+      dbPostgresSchema: 'views_popular',
+    })
+    close = server.close
+    agent = new AtpAgent({ service: server.url })
+    const pdsAgent = new AtpAgent({ service: server.pdsUrl })
+    sc = new SeedClient(pdsAgent)
+    await basicSeed(sc)
+    await sc.createAccount('eve', {
+      ...account,
+      email: '[email protected]',
+      handle: 'eve.test',
+      password: 'eve-pass',
+    })
+    await sc.createAccount('frank', {
+      ...account,
+      email: '[email protected]',
+      handle: 'frank.test',
+      password: 'frank-pass',
+    })
+    await processAll(server)
+    alice = sc.dids.alice
+    bob = sc.dids.bob
+    carol = sc.dids.carol
+    dan = sc.dids.dan
+    eve = sc.dids.eve
+    frank = sc.dids.frank
+  })
+
+  afterAll(async () => {
+    await close()
+  })
+
+  it('returns well liked posts', async () => {
+    const img = await sc.uploadFile(
+      alice,
+      'tests/image/fixtures/key-landscape-small.jpg',
+      'image/jpeg',
+    )
+    const one = await sc.post(alice, 'first post', undefined, [img])
+    await sc.like(bob, one.ref)
+    await sc.like(carol, one.ref)
+    await sc.like(dan, one.ref)
+    await sc.like(eve, one.ref)
+    await sc.like(frank, one.ref)
+    const two = await sc.post(bob, 'bobby boi')
+    await sc.like(alice, two.ref)
+    await sc.like(carol, two.ref)
+    await sc.like(dan, two.ref)
+    await sc.like(eve, two.ref)
+    await sc.like(frank, two.ref)
+    const three = await sc.reply(bob, one.ref, one.ref, 'reply')
+    await sc.like(alice, three.ref)
+    await sc.like(carol, three.ref)
+    await sc.like(dan, three.ref)
+    await sc.like(eve, three.ref)
+    await sc.like(frank, three.ref)
+    await processAll(server)
+
+    const res = await agent.api.app.bsky.unspecced.getPopular(
+      {},
+      { headers: sc.getHeaders(alice, true) },
+    )
+    const feedUris = res.data.feed.map((i) => i.post.uri).sort()
+    const expected = [one.ref.uriStr, two.ref.uriStr, three.ref.uriStr].sort()
+    expect(feedUris).toEqual(expected)
+  })
+})