merge main

lexicons/com/atproto/admin/defs.json

@@ -248,7 +248,8 @@
           }
         },
         "invitesDisabled": { "type": "boolean" },
-        "inviteNote": { "type": "string" }
+        "inviteNote": { "type": "string" },
+        "emailConfirmedAt": { "type": "string", "format": "datetime" }
       }
     },
     "accountView": {
@@ -271,6 +272,7 @@
           }
         },
         "invitesDisabled": { "type": "boolean" },
+        "emailConfirmedAt": { "type": "string", "format": "datetime" },
         "inviteNote": { "type": "string" }
       }
     },

packages/api/src/client/lexicons.ts

@@ -434,6 +434,10 @@ export const schemaDict = {
           inviteNote: {
             type: 'string',
           },
+          emailConfirmedAt: {
+            type: 'string',
+            format: 'datetime',
+          },
         },
       },
       accountView: {
@@ -469,6 +473,10 @@ export const schemaDict = {
           invitesDisabled: {
             type: 'boolean',
           },
+          emailConfirmedAt: {
+            type: 'string',
+            format: 'datetime',
+          },
           inviteNote: {
             type: 'string',
           },

packages/api/src/client/types/com/atproto/admin/defs.ts

@@ -241,6 +241,7 @@ export interface RepoViewDetail {
   invites?: ComAtprotoServerDefs.InviteCode[]
   invitesDisabled?: boolean
   inviteNote?: string
+  emailConfirmedAt?: string
   [k: string]: unknown
 }
 
@@ -264,6 +265,7 @@ export interface AccountView {
   invitedBy?: ComAtprotoServerDefs.InviteCode
   invites?: ComAtprotoServerDefs.InviteCode[]
   invitesDisabled?: boolean
+  emailConfirmedAt?: string
   inviteNote?: string
   [k: string]: unknown
 }

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

@@ -42,12 +42,12 @@ const skeleton = async (
   ctx: Context,
 ): Promise<SkeletonState> => {
   const { db } = ctx
-  const { limit, cursor, viewer } = params
+  const { viewer } = params
+  const alreadyIncluded = parseCursor(params.cursor)
   const { ref } = db.db.dynamic
-  let suggestionsQb = db.db
+  const suggestions = await db.db
     .selectFrom('suggested_follow')
     .innerJoin('actor', 'actor.did', 'suggested_follow.did')
-    .innerJoin('profile_agg', 'profile_agg.did', 'actor.did')
     .where(notSoftDeletedClause(ref('actor')))
     .where('suggested_follow.did', '!=', viewer ?? '')
     .whereNotExists((qb) =>
@@ -57,27 +57,30 @@ const skeleton = async (
         .where('creator', '=', viewer ?? '')
         .whereRef('subjectDid', '=', ref('actor.did')),
     )
+    .if(alreadyIncluded.length > 0, (qb) =>
+      qb.where('suggested_follow.order', 'not in', alreadyIncluded),
+    )
     .selectAll()
-    .select('profile_agg.postsCount as postsCount')
-    .limit(limit)
     .orderBy('suggested_follow.order', 'asc')
+    .execute()
 
-  if (cursor) {
-    const cursorRow = await db.db
-      .selectFrom('suggested_follow')
-      .where('did', '=', cursor)
-      .selectAll()
-      .executeTakeFirst()
-    if (cursorRow) {
-      suggestionsQb = suggestionsQb.where(
-        'suggested_follow.order',
-        '>',
-        cursorRow.order,
-      )
-    }
-  }
-  const suggestions = await suggestionsQb.execute()
-  return { params, suggestions, cursor: suggestions.at(-1)?.did }
+  // always include first two
+  const firstTwo = suggestions.filter(
+    (row) => row.order === 1 || row.order === 2,
+  )
+  const rest = suggestions.filter((row) => row.order !== 1 && row.order !== 2)
+  const limited = firstTwo.concat(shuffle(rest)).slice(0, params.limit)
+
+  // if the result set ends up getting larger, consider using a seed included in the cursor for for the randomized shuffle
+  const cursor =
+    limited.length > 0
+      ? limited
+          .map((row) => row.order.toString())
+          .concat(alreadyIncluded.map((id) => id.toString()))
+          .join(':')
+      : undefined
+
+  return { params, suggestions: limited, cursor }
 }
 
 const hydration = async (state: SkeletonState, ctx: Context) => {
@@ -110,6 +113,27 @@ const presentation = (state: HydrationState) => {
   return { actors: suggestedActors, cursor }
 }
 
+const parseCursor = (cursor?: string): number[] => {
+  if (!cursor) {
+    return []
+  }
+  try {
+    return cursor
+      .split(':')
+      .map((id) => parseInt(id, 10))
+      .filter((id) => !isNaN(id))
+  } catch {
+    return []
+  }
+}
+
+const shuffle = <T>(arr: T[]): T[] => {
+  return arr
+    .map((value) => ({ value, sort: Math.random() }))
+    .sort((a, b) => a.sort - b.sort)
+    .map(({ value }) => value)
+}
+
 type Context = {
   db: Database
   actorService: ActorService

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

@@ -0,0 +1,123 @@
+import AppContext from '../../../../context'
+import { Server } from '../../../../lexicon'
+import { InvalidRequestError } from '@atproto/xrpc-server'
+import AtpAgent from '@atproto/api'
+import { AtUri } from '@atproto/syntax'
+import { mapDefined } from '@atproto/common'
+import { QueryParams } from '../../../../lexicon/types/app/bsky/feed/searchPosts'
+import { Database } from '../../../../db'
+import { FeedHydrationState, FeedService } from '../../../../services/feed'
+import { ActorService } from '../../../../services/actor'
+import { createPipeline } from '../../../../pipeline'
+
+export default function (server: Server, ctx: AppContext) {
+  const searchPosts = createPipeline(
+    skeleton,
+    hydration,
+    noBlocks,
+    presentation,
+  )
+  server.app.bsky.feed.searchPosts({
+    auth: ctx.authOptionalVerifier,
+    handler: async ({ auth, params }) => {
+      const viewer = auth.credentials.did
+      const db = ctx.db.getReplica('search')
+      const feedService = ctx.services.feed(db)
+      const actorService = ctx.services.actor(db)
+      const searchAgent = ctx.searchAgent
+      if (!searchAgent) {
+        throw new InvalidRequestError('Search not available')
+      }
+
+      const results = await searchPosts(
+        { ...params, viewer },
+        { db, feedService, actorService, searchAgent },
+      )
+
+      return {
+        encoding: 'application/json',
+        body: results,
+      }
+    },
+  })
+}
+
+const skeleton = async (
+  params: Params,
+  ctx: Context,
+): Promise<SkeletonState> => {
+  const res = await ctx.searchAgent.api.app.bsky.unspecced.searchPostsSkeleton(
+    params,
+  )
+  return {
+    params,
+    postUris: res.data.posts.map((a) => a.uri),
+    cursor: res.data.cursor,
+  }
+}
+
+const hydration = async (
+  state: SkeletonState,
+  ctx: Context,
+): Promise<HydrationState> => {
+  const { feedService } = ctx
+  const { params, postUris } = state
+  const uris = new Set<string>(postUris)
+  const dids = new Set<string>(postUris.map((uri) => new AtUri(uri).hostname))
+  const hydrated = await feedService.feedHydration({
+    uris,
+    dids,
+    viewer: params.viewer,
+  })
+  return { ...state, ...hydrated }
+}
+
+const noBlocks = (state: HydrationState): HydrationState => {
+  const { viewer } = state.params
+  state.postUris = state.postUris.filter((uri) => {
+    const post = state.posts[uri]
+    if (!viewer || !post) return true
+    return !state.bam.block([viewer, post.creator])
+  })
+  return state
+}
+
+const presentation = (state: HydrationState, ctx: Context) => {
+  const { feedService, actorService } = ctx
+  const { postUris, profiles, params } = state
+  const actors = actorService.views.profileBasicPresentation(
+    Object.keys(profiles),
+    state,
+    { viewer: params.viewer },
+  )
+
+  const postViews = mapDefined(postUris, (uri) =>
+    feedService.views.formatPostView(
+      uri,
+      actors,
+      state.posts,
+      state.threadgates,
+      state.embeds,
+      state.labels,
+      state.lists,
+    ),
+  )
+  return { posts: postViews }
+}
+
+type Context = {
+  db: Database
+  feedService: FeedService
+  actorService: ActorService
+  searchAgent: AtpAgent
+}
+
+type Params = QueryParams & { viewer: string | null }
+
+type SkeletonState = {
+  params: Params
+  postUris: string[]
+  cursor?: string
+}
+
+type HydrationState = SkeletonState & FeedHydrationState

packages/bsky/src/api/com/atproto/admin/util.ts

@@ -32,6 +32,7 @@ export const addAccountInfoToRepoViewDetail = (
     invitesDisabled: accountInfo.invitesDisabled,
     inviteNote: accountInfo.inviteNote,
     invites: accountInfo.invites,
+    emailConfirmedAt: accountInfo.emailConfirmedAt,
   }
 }
 

packages/bsky/src/api/index.ts

@@ -13,6 +13,7 @@ import getLikes from './app/bsky/feed/getLikes'
 import getListFeed from './app/bsky/feed/getListFeed'
 import getPostThread from './app/bsky/feed/getPostThread'
 import getPosts from './app/bsky/feed/getPosts'
+import searchPosts from './app/bsky/feed/searchPosts'
 import getActorLikes from './app/bsky/feed/getActorLikes'
 import getProfile from './app/bsky/actor/getProfile'
 import getProfiles from './app/bsky/actor/getProfiles'
@@ -74,6 +75,7 @@ export default function (server: Server, ctx: AppContext) {
   getListFeed(server, ctx)
   getPostThread(server, ctx)
   getPosts(server, ctx)
+  searchPosts(server, ctx)
   getActorLikes(server, ctx)
   getProfile(server, ctx)
   getProfiles(server, ctx)

packages/bsky/src/lexicon/lexicons.ts

@@ -434,6 +434,10 @@ export const schemaDict = {
           inviteNote: {
             type: 'string',
           },
+          emailConfirmedAt: {
+            type: 'string',
+            format: 'datetime',
+          },
         },
       },
       accountView: {
@@ -469,6 +473,10 @@ export const schemaDict = {
           invitesDisabled: {
             type: 'boolean',
           },
+          emailConfirmedAt: {
+            type: 'string',
+            format: 'datetime',
+          },
           inviteNote: {
             type: 'string',
           },

packages/bsky/src/lexicon/types/com/atproto/admin/defs.ts

@@ -241,6 +241,7 @@ export interface RepoViewDetail {
   invites?: ComAtprotoServerDefs.InviteCode[]
   invitesDisabled?: boolean
   inviteNote?: string
+  emailConfirmedAt?: string
   [k: string]: unknown
 }
 
@@ -264,6 +265,7 @@ export interface AccountView {
   invitedBy?: ComAtprotoServerDefs.InviteCode
   invites?: ComAtprotoServerDefs.InviteCode[]
   invitesDisabled?: boolean
+  emailConfirmedAt?: string
   inviteNote?: string
   [k: string]: unknown
 }

packages/bsky/tests/admin/get-repo.test.ts

@@ -91,6 +91,39 @@ describe('admin get repo view', () => {
     expect(triage).toEqual({ ...admin, email: undefined })
   })
 
+  it('includes emailConfirmedAt timestamp', async () => {
+    const { data: beforeEmailVerification } =
+      await agent.api.com.atproto.admin.getRepo(
+        { did: sc.dids.bob },
+        { headers: network.pds.adminAuthHeaders() },
+      )
+
+    expect(beforeEmailVerification.emailConfirmedAt).toBeUndefined()
+    const timestampBeforeVerification = Date.now()
+    const bobsAccount = sc.accounts[sc.dids.bob]
+    const verificationToken = await network.pds.ctx.services
+      .account(network.pds.ctx.db)
+      .createEmailToken(sc.dids.bob, 'confirm_email')
+    await agent.api.com.atproto.server.confirmEmail(
+      { email: bobsAccount.email, token: verificationToken },
+      {
+        encoding: 'application/json',
+
+        headers: sc.getHeaders(sc.dids.bob),
+      },
+    )
+    const { data: afterEmailVerification } =
+      await agent.api.com.atproto.admin.getRepo(
+        { did: sc.dids.bob },
+        { headers: network.pds.adminAuthHeaders() },
+      )
+
+    expect(afterEmailVerification.emailConfirmedAt).toBeTruthy()
+    expect(
+      new Date(afterEmailVerification.emailConfirmedAt as string).getTime(),
+    ).toBeGreaterThan(timestampBeforeVerification)
+  })
+
   it('fails when repo does not exist.', async () => {
     const promise = agent.api.com.atproto.admin.getRepo(
       { did: 'did:plc:doesnotexist' },