implement getListMutes and -Blocks w/ dataplane

packages/bsky/src/api/app/bsky/graph/getListBlocks.ts

@@ -1,34 +1,29 @@
 import { mapDefined } from '@atproto/common'
 import { Server } from '../../../../lexicon'
 import { QueryParams } from '../../../../lexicon/types/app/bsky/graph/getListBlocks'
-import { paginate, TimeCidKeyset } from '../../../../db/pagination'
 import AppContext from '../../../../context'
-import { Database } from '../../../../db'
-import { Actor } from '../../../../db/tables/actor'
-import { GraphService, ListInfo } from '../../../../services/graph'
-import { ActorService, ProfileHydrationState } from '../../../../services/actor'
-import { createPipeline, noRules } from '../../../../pipeline'
+import {
+  createPipelineNew,
+  HydrationFnInput,
+  noRulesNew,
+  PresentationFnInput,
+  SkeletonFnInput,
+} from '../../../../pipeline'
+import { Hydrator } from '../../../../hydration/hydrator'
+import { Views } from '../../../../views'
 
 export default function (server: Server, ctx: AppContext) {
-  const getListBlocks = createPipeline(
+  const getListBlocks = createPipelineNew(
     skeleton,
     hydration,
-    noRules,
+    noRulesNew,
     presentation,
   )
   server.app.bsky.graph.getListBlocks({
     auth: ctx.authVerifier,
     handler: async ({ params, auth }) => {
-      const db = ctx.db.getReplica()
-      const graphService = ctx.services.graph(db)
-      const actorService = ctx.services.actor(db)
       const viewer = auth.credentials.did
-
-      const result = await getListBlocks(
-        { ...params, viewer },
-        { db, actorService, graphService },
-      )
-
+      const result = await getListBlocks({ ...params, viewer }, ctx)
       return {
         encoding: 'application/json',
         body: result,
@@ -38,78 +33,44 @@ export default function (server: Server, ctx: AppContext) {
 }
 
 const skeleton = async (
-  params: Params,
-  ctx: Context,
+  input: SkeletonFnInput<Context, Params>,
 ): Promise<SkeletonState> => {
-  const { db, graphService } = ctx
-  const { limit, cursor, viewer } = params
-  const { ref } = db.db.dynamic
-
-  let listsReq = graphService
-    .getListsQb(viewer)
-    .whereExists(
-      db.db
-        .selectFrom('list_block')
-        .where('list_block.creator', '=', viewer)
-        .whereRef('list_block.subjectUri', '=', ref('list.uri'))
-        .selectAll(),
-    )
-
-  const keyset = new TimeCidKeyset(ref('list.createdAt'), ref('list.cid'))
-
-  listsReq = paginate(listsReq, {
-    limit,
-    cursor,
-    keyset,
-  })
-
-  const listInfos = await listsReq.execute()
-
-  return {
-    params,
-    listInfos,
-    cursor: keyset.packFromResult(listInfos),
-  }
+  const { ctx, params } = input
+  const { listUris, cursor } =
+    await ctx.hydrator.dataplane.getBlocklistSubscriptions({
+      actorDid: params.viewer,
+      cursor: params.cursor,
+      limit: params.limit,
+    })
+  return { listUris, cursor }
 }
 
-const hydration = async (state: SkeletonState, ctx: Context) => {
-  const { actorService } = ctx
-  const { params, listInfos } = state
-  const profileState = await actorService.views.profileHydration(
-    listInfos.map((list) => list.creator),
-    { viewer: params.viewer },
-  )
-  return { ...state, ...profileState }
+const hydration = async (
+  input: HydrationFnInput<Context, Params, SkeletonState>,
+) => {
+  const { ctx, params, skeleton } = input
+  return await ctx.hydrator.hydrateLists(skeleton.listUris, params.viewer)
 }
 
-const presentation = (state: HydrationState, ctx: Context) => {
-  const { actorService, graphService } = ctx
-  const { params, listInfos, cursor, ...profileState } = state
-  const actors = actorService.views.profilePresentation(
-    Object.keys(profileState.profiles),
-    profileState,
-    params.viewer,
-  )
-  const lists = mapDefined(listInfos, (list) =>
-    graphService.formatListView(list, actors),
-  )
+const presentation = (
+  input: PresentationFnInput<Context, Params, SkeletonState>,
+) => {
+  const { ctx, skeleton, hydration } = input
+  const { listUris, cursor } = skeleton
+  const lists = mapDefined(listUris, (uri) => ctx.views.list(uri, hydration))
   return { lists, cursor }
 }
 
 type Context = {
-  db: Database
-  actorService: ActorService
-  graphService: GraphService
+  hydrator: Hydrator
+  views: Views
 }
 
 type Params = QueryParams & {
   viewer: string
 }
 
 type SkeletonState = {
-  params: Params
-  listInfos: (Actor & ListInfo)[]
+  listUris: string[]
   cursor?: string
 }
-
-type HydrationState = SkeletonState & ProfileHydrationState

packages/bsky/src/api/app/bsky/graph/getListMutes.ts

@@ -1,51 +1,76 @@
 import { mapDefined } from '@atproto/common'
 import { Server } from '../../../../lexicon'
-import { paginate, TimeCidKeyset } from '../../../../db/pagination'
+import { QueryParams } from '../../../../lexicon/types/app/bsky/graph/getListBlocks'
 import AppContext from '../../../../context'
+import {
+  createPipelineNew,
+  HydrationFnInput,
+  noRulesNew,
+  PresentationFnInput,
+  SkeletonFnInput,
+} from '../../../../pipeline'
+import { Hydrator } from '../../../../hydration/hydrator'
+import { Views } from '../../../../views'
 
 export default function (server: Server, ctx: AppContext) {
+  const getListMutes = createPipelineNew(
+    skeleton,
+    hydration,
+    noRulesNew,
+    presentation,
+  )
   server.app.bsky.graph.getListMutes({
     auth: ctx.authVerifier,
     handler: async ({ params, auth }) => {
-      const { limit, cursor } = params
-      const requester = auth.credentials.did
-      const db = ctx.db.getReplica()
-      const { ref } = db.db.dynamic
+      const viewer = auth.credentials.did
+      const result = await getListMutes({ ...params, viewer }, ctx)
+      return {
+        encoding: 'application/json',
+        body: result,
+      }
+    },
+  })
+}
 
-      const graphService = ctx.services.graph(db)
+const skeleton = async (
+  input: SkeletonFnInput<Context, Params>,
+): Promise<SkeletonState> => {
+  const { ctx, params } = input
+  const { listUris, cursor } =
+    await ctx.hydrator.dataplane.getMutelistSubscriptions({
+      actorDid: params.viewer,
+      cursor: params.cursor,
+      limit: params.limit,
+    })
+  return { listUris, cursor }
+}
 
-      let listsReq = graphService
-        .getListsQb(requester)
-        .whereExists(
-          db.db
-            .selectFrom('list_mute')
-            .where('list_mute.mutedByDid', '=', requester)
-            .whereRef('list_mute.listUri', '=', ref('list.uri'))
-            .selectAll(),
-        )
+const hydration = async (
+  input: HydrationFnInput<Context, Params, SkeletonState>,
+) => {
+  const { ctx, params, skeleton } = input
+  return await ctx.hydrator.hydrateLists(skeleton.listUris, params.viewer)
+}
 
-      const keyset = new TimeCidKeyset(ref('list.createdAt'), ref('list.cid'))
-      listsReq = paginate(listsReq, {
-        limit,
-        cursor,
-        keyset,
-      })
-      const listsRes = await listsReq.execute()
+const presentation = (
+  input: PresentationFnInput<Context, Params, SkeletonState>,
+) => {
+  const { ctx, skeleton, hydration } = input
+  const { listUris, cursor } = skeleton
+  const lists = mapDefined(listUris, (uri) => ctx.views.list(uri, hydration))
+  return { lists, cursor }
+}
 
-      const actorService = ctx.services.actor(db)
-      const profiles = await actorService.views.profiles(listsRes, requester)
+type Context = {
+  hydrator: Hydrator
+  views: Views
+}
 
-      const lists = mapDefined(listsRes, (row) =>
-        graphService.formatListView(row, profiles),
-      )
+type Params = QueryParams & {
+  viewer: string
+}
 
-      return {
-        encoding: 'application/json',
-        body: {
-          lists,
-          cursor: keyset.packFromResult(listsRes),
-        },
-      }
-    },
-  })
+type SkeletonState = {
+  listUris: string[]
+  cursor?: string
 }

packages/bsky/src/hydration/graph.ts

@@ -95,7 +95,7 @@ export class GraphHydrator {
     return uris.reduce((acc, uri, i) => {
       return acc.set(uri, {
         viewerMuted: mutesAndBlocks[i].muted ? uri : undefined,
-        viewerListBlockUri: mutesAndBlocks[i].listBlockUri,
+        viewerListBlockUri: mutesAndBlocks[i].listBlockUri || undefined,
         viewerInList: listMemberships.listitemUris[i],
       })
     }, new HydrationMap<ListViewerState>())

packages/bsky/src/views/index.ts

@@ -155,7 +155,7 @@ export class Views {
     const creatorDid = new AtUri(uri).hostname
     const list = state.lists?.get(uri)
     if (!list) return
-    const creator = this.profileBasic(creatorDid, state)
+    const creator = this.profile(creatorDid, state)
     if (!creator) return
     const basicView = this.listBasic(uri, state)
     if (!basicView) return

packages/bsky/tests/views/mute-lists.test.ts

@@ -93,7 +93,6 @@ describe('bsky views with mutes from mute lists', () => {
   })
 
   it('uses a list for mutes', async () => {
-    // @TODO proxy through appview
     await agent.api.app.bsky.graph.muteActorList(
       {
         list: listUri,